aether_cli/acp/
mappers.rs

1use acp_utils::notifications::{ContextClearedParams, ContextUsageParams, SubAgentProgressParams};
2use acp_utils::server::AcpActorHandle;
3use aether_core::events::{AgentMessage, SubAgentProgressPayload};
4use agent_client_protocol::{
5    self as acp, Content, ContentBlock, ContentChunk, Diff, HttpHeader, McpServer, PlanEntry, PlanEntryPriority,
6    PlanEntryStatus, SessionId, SessionNotification, SessionUpdate, StopReason, TextContent, ToolCall, ToolCallContent,
7    ToolCallId, ToolCallStatus, ToolCallUpdate, ToolCallUpdateFields,
8};
9use llm::{ToolCallError, ToolCallRequest, ToolCallResult};
10use mcp_utils::client::{McpServerConfig, ServerConfig};
11use mcp_utils::display_meta::{PlanMetaStatus, ToolResultMeta};
12use rmcp::model::Prompt as McpPrompt;
13use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig;
14
15use aether_core::context::ext::{SessionEvent, UserEvent};
16
17/// Converts an MCP Prompt to an ACP `AvailableCommand`
18///
19/// Strips the MCP namespace from the prompt name (e.g., "`coding__web`" -> "web")
20/// and creates a slash command that clients can invoke.
21pub fn map_mcp_prompt_to_available_command(prompt: &McpPrompt) -> acp::AvailableCommand {
22    // Extract the base command name by removing the namespace prefix
23    let command_name = prompt.name.split("__").last().unwrap_or(prompt.name.as_ref()).to_string();
24
25    // Extract the input hint from the unified prompt format's ARGUMENTS parameter,
26    // falling back to a generic hint.
27    let hint = prompt
28        .arguments
29        .as_ref()
30        .and_then(|args| args.iter().find(|a| a.name.as_str() == "ARGUMENTS").and_then(|a| a.description.as_deref()))
31        .unwrap_or("optional arguments");
32    let input = Some(acp::AvailableCommandInput::Unstructured(acp::UnstructuredCommandInput::new(hint)));
33
34    let description = prompt.description.clone().unwrap_or_else(|| "No description available".to_string());
35
36    acp::AvailableCommand::new(command_name, description).input(input)
37}
38
39/// Maps ACP MCP server definitions to internal `McpServerConfig`, skipping unsupported transports.
40pub fn map_acp_mcp_servers(servers: Vec<McpServer>) -> Vec<McpServerConfig> {
41    servers
42        .into_iter()
43        .filter_map(|s| {
44            try_map_mcp_server(s).or_else(|| {
45                tracing::warn!("Unsupported ACP MCP server transport, skipping");
46                None
47            })
48        })
49        .collect()
50}
51
52fn try_map_mcp_server(server: McpServer) -> Option<McpServerConfig> {
53    use McpServer::{Http, Sse, Stdio};
54    match server {
55        Stdio(stdio) => Some(
56            ServerConfig::Stdio {
57                name: stdio.name,
58                command: stdio.command.to_string_lossy().into_owned(),
59                args: stdio.args,
60                env: stdio.env.into_iter().map(|e| (e.name, e.value)).collect(),
61            }
62            .into(),
63        ),
64
65        Http(http) => Some(ServerConfig::Http { name: http.name, config: http_config(http.url, &http.headers) }.into()),
66
67        Sse(sse) => Some(ServerConfig::Http { name: sse.name, config: http_config(sse.url, &sse.headers) }.into()),
68
69        _ => None,
70    }
71}
72
73fn http_config(url: String, headers: &[HttpHeader]) -> StreamableHttpClientTransportConfig {
74    let auth_header = headers.iter().find(|h| h.name.eq_ignore_ascii_case("authorization")).map(|h| h.value.clone());
75
76    let mut config = StreamableHttpClientTransportConfig::with_uri(url);
77    if let Some(auth) = auth_header {
78        config = config.auth_header(auth);
79    }
80    config
81}
82
83/// Converts Aether `AgentMessage` to ACP `SessionUpdate`
84pub fn map_agent_message_to_session_notification(
85    session_id: SessionId,
86    msg: &AgentMessage,
87) -> Option<SessionNotification> {
88    map_agent_message_to_notification(session_id, msg, NotificationMode::Live)
89}
90
91#[derive(Clone, Copy)]
92enum NotificationMode {
93    Live,
94    Replay,
95}
96
97fn map_agent_message_to_notification(
98    session_id: SessionId,
99    msg: &AgentMessage,
100    mode: NotificationMode,
101) -> Option<SessionNotification> {
102    match msg {
103        AgentMessage::Text { chunk, is_complete, .. } => {
104            map_chunk_to_notification(session_id, chunk, *is_complete, mode, SessionUpdate::AgentMessageChunk)
105        }
106
107        AgentMessage::Thought { chunk, is_complete, .. } => {
108            map_chunk_to_notification(session_id, chunk, *is_complete, mode, SessionUpdate::AgentThoughtChunk)
109        }
110
111        AgentMessage::ToolCall { request, .. } => Some(map_tool_call_to_notification(session_id, request)),
112
113        AgentMessage::ToolCallUpdate { tool_call_id, chunk, .. } => {
114            Some(map_tool_call_update_to_notification(session_id, tool_call_id, chunk))
115        }
116
117        AgentMessage::ToolResult { result, result_meta, .. } => {
118            Some(map_tool_result_to_notification(session_id, result, result_meta.as_ref()))
119        }
120
121        AgentMessage::ToolError { error, .. } => Some(map_tool_error_to_notification(session_id, error)),
122
123        AgentMessage::ToolProgress { request, progress, total, message } => {
124            map_tool_progress_to_notification(session_id, request, *progress, *total, message.as_ref())
125        }
126
127        AgentMessage::Error { message } => Some(acp::SessionNotification::new(
128            session_id,
129            SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::Text(TextContent::new(format!(
130                "[Error] {message}"
131            ))))),
132        )),
133
134        AgentMessage::ContextUsageUpdate { .. }
135        | AgentMessage::ContextCleared
136        | AgentMessage::Cancelled { .. }
137        | AgentMessage::Done
138        | AgentMessage::ContextCompactionStarted { .. }
139        | AgentMessage::ContextCompactionResult { .. }
140        | AgentMessage::AutoContinue { .. }
141        | AgentMessage::ModelSwitched { .. } => None,
142    }
143}
144
145pub fn try_into_ext_notification(msg: &AgentMessage) -> Option<acp::ExtNotification> {
146    match msg {
147        AgentMessage::ContextUsageUpdate {
148            usage_ratio,
149            context_limit,
150            input_tokens,
151            output_tokens,
152            cache_read_tokens,
153            cache_creation_tokens,
154            reasoning_tokens,
155            total_input_tokens,
156            total_output_tokens,
157            total_cache_read_tokens,
158            total_cache_creation_tokens,
159            total_reasoning_tokens,
160        } => {
161            let params = ContextUsageParams {
162                usage_ratio: *usage_ratio,
163                context_limit: *context_limit,
164                input_tokens: *input_tokens,
165                output_tokens: *output_tokens,
166                cache_read_tokens: *cache_read_tokens,
167                cache_creation_tokens: *cache_creation_tokens,
168                reasoning_tokens: *reasoning_tokens,
169                total_input_tokens: *total_input_tokens,
170                total_output_tokens: *total_output_tokens,
171                total_cache_read_tokens: *total_cache_read_tokens,
172                total_cache_creation_tokens: *total_cache_creation_tokens,
173                total_reasoning_tokens: *total_reasoning_tokens,
174            };
175            Some(params.into())
176        }
177        AgentMessage::ToolProgress { request, message, .. } => {
178            let msg_str = message.as_ref()?;
179            let params = try_parse_sub_agent_progress(msg_str, request)?;
180            Some(params.into())
181        }
182        AgentMessage::ContextCleared => Some(ContextClearedParams::default().into()),
183        _ => None,
184    }
185}
186
187/// If the tool result carries plan metadata, build a `SessionUpdate::Plan` notification.
188pub fn try_extract_plan_notification(
189    session_id: SessionId,
190    result_meta: Option<&ToolResultMeta>,
191) -> Option<SessionNotification> {
192    let plan_meta = result_meta?.plan.as_ref()?;
193    let entries = plan_meta
194        .entries
195        .iter()
196        .map(|e| PlanEntry::new(e.content.clone(), PlanEntryPriority::Medium, plan_status_to_acp(e.status)))
197        .collect();
198    Some(SessionNotification::new(session_id, SessionUpdate::Plan(acp::Plan::new(entries))))
199}
200
201/// Convert internal plan status to ACP protocol status.
202fn plan_status_to_acp(status: PlanMetaStatus) -> PlanEntryStatus {
203    match status {
204        PlanMetaStatus::InProgress => PlanEntryStatus::InProgress,
205        PlanMetaStatus::Completed => PlanEntryStatus::Completed,
206        PlanMetaStatus::Pending => PlanEntryStatus::Pending,
207    }
208}
209
210/// Determines the stop reason from the final agent message
211pub fn map_agent_message_to_stop_reason(msg: &AgentMessage) -> acp::StopReason {
212    match msg {
213        AgentMessage::Cancelled { .. } => StopReason::Cancelled,
214        _ => StopReason::EndTurn,
215    }
216}
217
218fn map_chunk_to_notification(
219    session_id: SessionId,
220    chunk: &str,
221    is_complete: bool,
222    mode: NotificationMode,
223    wrap: fn(ContentChunk) -> SessionUpdate,
224) -> Option<SessionNotification> {
225    match mode {
226        // Skip the final completion message to avoid sending duplicate content.
227        // The client has already received all the chunks during streaming.
228        NotificationMode::Live if is_complete => return None,
229        NotificationMode::Replay if !is_complete => return None,
230        NotificationMode::Live | NotificationMode::Replay => {}
231    }
232
233    Some(acp::SessionNotification::new(
234        session_id,
235        wrap(ContentChunk::new(ContentBlock::Text(TextContent::new(chunk.to_owned())))),
236    ))
237}
238
239fn map_tool_call_to_notification(session_id: SessionId, request: &ToolCallRequest) -> SessionNotification {
240    let raw_input = serde_json::from_str(&request.arguments).ok();
241    SessionNotification::new(
242        session_id,
243        SessionUpdate::ToolCall(
244            ToolCall::new(ToolCallId::new(request.id.clone()), humanize_tool_name(&request.name))
245                .status(acp::ToolCallStatus::InProgress)
246                .raw_input(raw_input),
247        ),
248    )
249}
250
251fn parse_tool_call_chunk(chunk: &str) -> serde_json::Value {
252    serde_json::from_str(chunk).unwrap_or_else(|_| serde_json::Value::String(chunk.to_string()))
253}
254
255fn map_tool_call_update_to_notification(session_id: SessionId, tool_call_id: &str, chunk: &str) -> SessionNotification {
256    let fields = ToolCallUpdateFields::new().status(ToolCallStatus::InProgress).raw_input(parse_tool_call_chunk(chunk));
257
258    SessionNotification::new(
259        session_id,
260        SessionUpdate::ToolCallUpdate(ToolCallUpdate::new(ToolCallId::new(tool_call_id.to_string()), fields)),
261    )
262}
263
264/// Produces the initial human-readable title for a tool call (e.g., "Read file").
265/// This is sent when the tool call starts.
266fn humanize_tool_name(name: &str) -> String {
267    let base = name.split("__").last().unwrap_or(name);
268    let mut result = base.replace('_', " ");
269    if let Some(first) = result.get_mut(0..1) {
270        first.make_ascii_uppercase();
271    }
272    result
273}
274
275fn map_tool_result_to_notification(
276    session_id: SessionId,
277    result: &ToolCallResult,
278    result_meta: Option<&ToolResultMeta>,
279) -> SessionNotification {
280    let mut content =
281        vec![ToolCallContent::Content(Content::new(ContentBlock::Text(TextContent::new(result.result.clone()))))];
282
283    if let Some(rm) = result_meta
284        && let Some(fd) = &rm.file_diff
285    {
286        let mut diff = Diff::new(&fd.path, &fd.new_text);
287        if let Some(old) = &fd.old_text {
288            diff = diff.old_text(old.clone());
289        }
290        content.push(ToolCallContent::Diff(diff));
291    }
292
293    let mut fields = ToolCallUpdateFields::new().status(ToolCallStatus::Completed).content(content);
294
295    if let Some(rm) = result_meta {
296        fields = fields.title(&rm.display.title);
297    }
298
299    let mut update = ToolCallUpdate::new(ToolCallId::new(result.id.clone()), fields);
300
301    if let Some(rm) = result_meta
302        && !rm.display.value.is_empty()
303    {
304        let mut meta_map = serde_json::Map::new();
305        meta_map.insert("display_value".into(), rm.display.value.clone().into());
306        update = update.meta(meta_map);
307    }
308
309    SessionNotification::new(session_id, SessionUpdate::ToolCallUpdate(update))
310}
311
312fn map_tool_error_to_notification(session_id: SessionId, error: &ToolCallError) -> SessionNotification {
313    SessionNotification::new(
314        session_id,
315        SessionUpdate::ToolCallUpdate(ToolCallUpdate::new(
316            ToolCallId::new(error.id.clone()),
317            ToolCallUpdateFields::new().status(ToolCallStatus::Failed).content(vec![ToolCallContent::Content(
318                Content::new(ContentBlock::Text(TextContent::new(error.error.clone()))),
319            )]),
320        )),
321    )
322}
323
324fn map_tool_progress_to_notification(
325    session_id: SessionId,
326    request: &ToolCallRequest,
327    progress: f64,
328    total: Option<f64>,
329    message: Option<&String>,
330) -> Option<SessionNotification> {
331    tracing::info!("Tool progress: {message:?}");
332
333    if message.and_then(|msg_str| try_parse_sub_agent_progress(msg_str, request)).is_some() {
334        return None;
335    }
336
337    if let Some(result_meta) = message.and_then(|m| try_parse_display_meta(m)) {
338        let fields = ToolCallUpdateFields::new().status(ToolCallStatus::InProgress).title(&result_meta.display.title);
339
340        let mut update = ToolCallUpdate::new(ToolCallId::new(request.id.clone()), fields);
341
342        if !result_meta.display.value.is_empty() {
343            let mut meta_map = serde_json::Map::new();
344            meta_map.insert("display_value".into(), result_meta.display.value.into());
345            update = update.meta(meta_map);
346        }
347
348        return Some(SessionNotification::new(session_id, SessionUpdate::ToolCallUpdate(update)));
349    }
350
351    let total_str = total.map_or_else(|| "?".to_string(), |t| t.to_string());
352    let progress_text = message
353        .map_or_else(|| format!("Progress: {progress}/{total_str}"), |msg| format!("{msg} ({progress}/{total_str})"));
354
355    Some(SessionNotification::new(
356        session_id,
357        SessionUpdate::ToolCallUpdate(ToolCallUpdate::new(
358            ToolCallId::new(request.id.clone()),
359            ToolCallUpdateFields::new().status(ToolCallStatus::InProgress).content(vec![ToolCallContent::Content(
360                Content::new(ContentBlock::Text(TextContent::new(progress_text))),
361            )]),
362        )),
363    ))
364}
365
366/// Replay session events to the client as ACP notifications.
367pub async fn replay_to_client(events: &[SessionEvent], actor_handle: &AcpActorHandle, session_id: &SessionId) {
368    for event in events {
369        let notifications: Vec<_> = match event {
370            SessionEvent::User(UserEvent::Message { content }) => content
371                .iter()
372                .map(|block| {
373                    SessionNotification::new(
374                        session_id.clone(),
375                        SessionUpdate::UserMessageChunk(ContentChunk::new(map_user_content_block(block))),
376                    )
377                })
378                .collect(),
379            SessionEvent::Agent(message) => {
380                map_agent_message_to_notification(session_id.clone(), message, NotificationMode::Replay)
381                    .into_iter()
382                    .collect()
383            }
384            SessionEvent::User(_) => Vec::new(),
385        };
386
387        for notif in notifications {
388            if let Err(e) = actor_handle.send_session_notification(notif).await {
389                tracing::error!("Failed to send replay notification: {e:?}");
390            }
391        }
392    }
393}
394
395fn map_user_content_block(block: &llm::ContentBlock) -> ContentBlock {
396    match block {
397        llm::ContentBlock::Text { text } => ContentBlock::Text(TextContent::new(text.clone())),
398        llm::ContentBlock::Image { data, mime_type } => {
399            ContentBlock::Image(acp::ImageContent::new(data.clone(), mime_type.clone()))
400        }
401        llm::ContentBlock::Audio { data, mime_type } => {
402            ContentBlock::Audio(acp::AudioContent::new(data.clone(), mime_type.clone()))
403        }
404    }
405}
406
407fn try_parse_display_meta(message: &str) -> Option<ToolResultMeta> {
408    serde_json::from_str::<ToolResultMeta>(message).ok()
409}
410
411/// Attempt to parse a tool progress message as sub-agent progress.
412fn try_parse_sub_agent_progress(message: &str, request: &llm::ToolCallRequest) -> Option<SubAgentProgressParams> {
413    let payload: SubAgentProgressPayload = serde_json::from_str(message).ok()?;
414
415    Some(SubAgentProgressParams {
416        parent_tool_id: request.id.clone(),
417        task_id: payload.task_id,
418        agent_name: payload.agent_name,
419        event: (&payload.event).into(),
420    })
421}
422
423#[cfg(test)]
424mod tests {
425    use super::*;
426    use acp_utils::notifications::SubAgentEvent;
427    use acp_utils::server::AcpRequest;
428    use aether_core::events::SUB_AGENT_PROGRESS_METHOD;
429    use llm::ToolCallRequest;
430    use tokio::sync::mpsc;
431
432    #[test]
433    fn test_tool_progress_with_sub_agent_payload_emits_ext_notification() {
434        let session_id = acp::SessionId::new("test-session");
435
436        let payload = SubAgentProgressPayload {
437            task_id: "task_1".to_string(),
438            agent_name: "sub-agent".to_string(),
439            event: AgentMessage::Text {
440                message_id: "msg_1".to_string(),
441                chunk: "Hello".to_string(),
442                is_complete: false,
443                model_name: "TestModel".to_string(),
444            },
445        };
446        let serialized_msg = serde_json::to_string(&payload).unwrap();
447
448        let tool_progress = AgentMessage::ToolProgress {
449            request: ToolCallRequest {
450                id: "call_123".to_string(),
451                name: "plugins__spawn_subagent".to_string(),
452                arguments: "{}".to_string(),
453            },
454            progress: 42.0,
455            total: Some(100.0),
456            message: Some(serialized_msg.clone()),
457        };
458
459        let notification = map_agent_message_to_session_notification(session_id.clone(), &tool_progress);
460
461        assert!(notification.is_none());
462
463        let ext = try_into_ext_notification(&tool_progress).expect("ext notification");
464        assert_eq!(ext.method.as_ref(), SUB_AGENT_PROGRESS_METHOD);
465
466        let parsed: SubAgentProgressParams = serde_json::from_str(ext.params.get()).expect("valid JSON");
467        assert_eq!(parsed.parent_tool_id, "call_123");
468        assert_eq!(parsed.task_id, "task_1");
469        assert_eq!(parsed.agent_name, "sub-agent");
470        assert!(matches!(parsed.event, SubAgentEvent::Other));
471    }
472
473    #[tokio::test]
474    async fn replay_emits_user_media_chunks_in_order() {
475        let (tx, mut rx) = mpsc::unbounded_channel();
476        let actor = AcpActorHandle::new(tx);
477        let session_id = acp::SessionId::new("test-session");
478        let events = vec![SessionEvent::User(UserEvent::Message {
479            content: vec![
480                llm::ContentBlock::text("hello"),
481                llm::ContentBlock::Image { data: "aW1n".to_string(), mime_type: "image/png".to_string() },
482                llm::ContentBlock::Audio { data: "YXVkaW8=".to_string(), mime_type: "audio/wav".to_string() },
483            ],
484        })];
485
486        let responder = tokio::spawn(async move {
487            let mut updates = Vec::new();
488            for _ in 0..3 {
489                if let Some(AcpRequest::SessionNotification { notification, response_tx }) = rx.recv().await {
490                    updates.push(notification.update);
491                    let _ = response_tx.send(Ok(()));
492                }
493            }
494            updates
495        });
496
497        replay_to_client(&events, &actor, &session_id).await;
498
499        let updates = responder.await.unwrap();
500        assert!(matches!(
501            &updates[0],
502            acp::SessionUpdate::UserMessageChunk(chunk)
503                if matches!(&chunk.content, acp::ContentBlock::Text(text) if text.text == "hello")
504        ));
505        assert!(matches!(
506            &updates[1],
507            acp::SessionUpdate::UserMessageChunk(chunk)
508                if matches!(&chunk.content, acp::ContentBlock::Image(_))
509        ));
510        assert!(matches!(
511            &updates[2],
512            acp::SessionUpdate::UserMessageChunk(chunk)
513                if matches!(&chunk.content, acp::ContentBlock::Audio(_))
514        ));
515    }
516
517    #[test]
518    fn test_thought_maps_to_agent_thought_chunk() {
519        let session_id = acp::SessionId::new("test-session");
520        let thought = AgentMessage::Thought {
521            message_id: "msg_1".to_string(),
522            chunk: "thinking...".to_string(),
523            is_complete: false,
524            model_name: "TestModel".to_string(),
525        };
526
527        let notification = map_agent_message_to_session_notification(session_id, &thought).expect("notification");
528
529        match notification.update {
530            acp::SessionUpdate::AgentThoughtChunk(chunk) => match chunk.content {
531                acp::ContentBlock::Text(text) => assert_eq!(text.text, "thinking..."),
532                other => panic!("Expected text content, got {other:?}"),
533            },
534            other => panic!("Expected AgentThoughtChunk, got {other:?}"),
535        }
536    }
537
538    #[test]
539    fn test_tool_call_maps_to_tool_call_notification() {
540        let session_id = acp::SessionId::new("test-session");
541        let message = AgentMessage::ToolCall {
542            request: ToolCallRequest {
543                id: "call_1".to_string(),
544                name: "coding__read_file".to_string(),
545                arguments: "{}".to_string(),
546            },
547            model_name: "TestModel".to_string(),
548        };
549
550        let notification = map_agent_message_to_session_notification(session_id, &message).expect("notification");
551
552        match notification.update {
553            acp::SessionUpdate::ToolCall(tool_call) => {
554                assert_eq!(tool_call.tool_call_id.0.as_ref(), "call_1");
555                assert_eq!(tool_call.title, "Read file");
556                assert_eq!(tool_call.status, acp::ToolCallStatus::InProgress);
557            }
558            other => panic!("Expected ToolCall, got {other:?}"),
559        }
560    }
561
562    #[test]
563    fn test_tool_call_update_maps_to_tool_call_update_notification() {
564        let session_id = acp::SessionId::new("test-session");
565        let message = AgentMessage::ToolCallUpdate {
566            tool_call_id: "call_1".to_string(),
567            chunk: r#"{"filePath":"Cargo.toml"}"#.to_string(),
568            model_name: "TestModel".to_string(),
569        };
570
571        let notification = map_agent_message_to_session_notification(session_id, &message).expect("notification");
572
573        match notification.update {
574            acp::SessionUpdate::ToolCallUpdate(update) => {
575                assert_eq!(update.tool_call_id.0.as_ref(), "call_1");
576                assert_eq!(update.fields.status, Some(acp::ToolCallStatus::InProgress));
577                assert_eq!(update.fields.raw_input, Some(serde_json::json!({ "filePath": "Cargo.toml" })));
578            }
579            other => panic!("Expected ToolCallUpdate, got {other:?}"),
580        }
581    }
582
583    #[test]
584    fn test_tool_call_update_has_same_live_and_replay_mapping() {
585        let session_id = acp::SessionId::new("test-session");
586        let message = AgentMessage::ToolCallUpdate {
587            tool_call_id: "call_1".to_string(),
588            chunk: r#"{"filePath":"Cargo.toml"}"#.to_string(),
589            model_name: "TestModel".to_string(),
590        };
591
592        let live = map_agent_message_to_notification(session_id.clone(), &message, NotificationMode::Live)
593            .expect("live notification");
594        let replay = map_agent_message_to_notification(session_id, &message, NotificationMode::Replay)
595            .expect("replay notification");
596
597        match (live.update, replay.update) {
598            (acp::SessionUpdate::ToolCallUpdate(live), acp::SessionUpdate::ToolCallUpdate(replay)) => {
599                assert_eq!(live.tool_call_id.0, replay.tool_call_id.0);
600                assert_eq!(live.fields.status, replay.fields.status);
601                assert_eq!(live.fields.raw_input, replay.fields.raw_input);
602            }
603            other => panic!("Expected ToolCallUpdate pair, got {other:?}"),
604        }
605    }
606
607    #[test]
608    fn test_live_mapping_skips_completed_chunks_but_replay_keeps_them() {
609        let cases: Vec<(AgentMessage, &str)> = vec![
610            (
611                AgentMessage::Text {
612                    message_id: "msg_1".to_string(),
613                    chunk: "done".to_string(),
614                    is_complete: true,
615                    model_name: "TestModel".to_string(),
616                },
617                "done",
618            ),
619            (
620                AgentMessage::Thought {
621                    message_id: "msg_1".to_string(),
622                    chunk: "final reasoning".to_string(),
623                    is_complete: true,
624                    model_name: "TestModel".to_string(),
625                },
626                "final reasoning",
627            ),
628        ];
629
630        for (message, expected_text) in cases {
631            let session_id = acp::SessionId::new("test-session");
632            assert!(
633                map_agent_message_to_notification(session_id.clone(), &message, NotificationMode::Live).is_none(),
634                "live mode should skip completed chunk"
635            );
636
637            let notification = map_agent_message_to_notification(session_id, &message, NotificationMode::Replay)
638                .expect("replay notification");
639
640            match notification.update {
641                acp::SessionUpdate::AgentMessageChunk(chunk) | acp::SessionUpdate::AgentThoughtChunk(chunk) => {
642                    match chunk.content {
643                        acp::ContentBlock::Text(text) => assert_eq!(text.text, expected_text),
644                        other => panic!("Expected text content, got {other:?}"),
645                    }
646                }
647                other => panic!("Expected chunk update, got {other:?}"),
648            }
649        }
650    }
651
652    #[test]
653    fn test_context_cleared_maps_to_ext_notification() {
654        let ext = try_into_ext_notification(&AgentMessage::ContextCleared)
655            .expect("context cleared should emit ext notification");
656        assert_eq!(ext.method.as_ref(), acp_utils::notifications::CONTEXT_CLEARED_METHOD);
657
658        let parsed: acp_utils::notifications::ContextClearedParams =
659            serde_json::from_str(ext.params.get()).expect("valid JSON");
660        assert_eq!(parsed, acp_utils::notifications::ContextClearedParams::default());
661    }
662
663    #[test]
664    fn test_tool_progress_with_invalid_json_falls_back_to_simple_message() {
665        let session_id = acp::SessionId::new("test-session");
666
667        // Simulate a tool progress message with invalid JSON
668        let tool_progress = AgentMessage::ToolProgress {
669            request: ToolCallRequest {
670                id: "call_456".to_string(),
671                name: "some_tool".to_string(),
672                arguments: "{}".to_string(),
673            },
674            progress: 50.0,
675            total: None,
676            message: Some("not valid json".to_string()),
677        };
678
679        let notification = map_agent_message_to_session_notification(session_id.clone(), &tool_progress);
680
681        assert!(notification.is_some());
682
683        // Should still produce a notification with the message as-is
684        let notification = notification.unwrap();
685        match notification.update {
686            acp::SessionUpdate::ToolCallUpdate(update) => {
687                if let Some(content) = &update.fields.content
688                    && let acp::ToolCallContent::Content(c) = &content[0]
689                    && let acp::ContentBlock::Text(text) = &c.content
690                {
691                    // Should contain the original message
692                    assert!(text.text.contains("not valid json"));
693                }
694            }
695            _ => panic!("Expected ToolCallUpdate"),
696        }
697    }
698
699    #[test]
700    fn test_map_acp_stdio_server() {
701        let server = acp::McpServer::Stdio(
702            acp::McpServerStdio::new("my-server", "/usr/bin/server")
703                .args(vec!["--port".into(), "8080".into()])
704                .env(vec![acp::EnvVariable::new("FOO", "bar")]),
705        );
706
707        let configs = map_acp_mcp_servers(vec![server]);
708        assert_eq!(configs.len(), 1);
709
710        match &configs[0] {
711            McpServerConfig::Server(ServerConfig::Stdio { name, command, args, env }) => {
712                assert_eq!(name, "my-server");
713                assert_eq!(command, "/usr/bin/server");
714                assert_eq!(args, &["--port", "8080"]);
715                assert_eq!(env.get("FOO").unwrap(), "bar");
716            }
717            other => panic!("Expected Stdio, got {other:?}"),
718        }
719    }
720
721    #[test]
722    fn test_map_acp_http_server() {
723        let server = acp::McpServer::Http(
724            acp::McpServerHttp::new("http-server", "https://example.com/mcp")
725                .headers(vec![acp::HttpHeader::new("Authorization", "Bearer token123")]),
726        );
727
728        let configs = map_acp_mcp_servers(vec![server]);
729        assert_eq!(configs.len(), 1);
730
731        match &configs[0] {
732            McpServerConfig::Server(ServerConfig::Http { name, config }) => {
733                assert_eq!(name, "http-server");
734                assert_eq!(config.uri.as_ref(), "https://example.com/mcp");
735                assert_eq!(config.auth_header.as_deref(), Some("Bearer token123"));
736            }
737            other => panic!("Expected Http, got {other:?}"),
738        }
739    }
740
741    #[test]
742    fn test_map_acp_sse_server() {
743        let server = acp::McpServer::Sse(acp::McpServerSse::new("sse-server", "https://example.com/sse"));
744
745        let configs = map_acp_mcp_servers(vec![server]);
746        assert_eq!(configs.len(), 1);
747
748        match &configs[0] {
749            McpServerConfig::Server(ServerConfig::Http { name, config }) => {
750                assert_eq!(name, "sse-server");
751                assert_eq!(config.uri.as_ref(), "https://example.com/sse");
752                assert_eq!(config.auth_header, None);
753            }
754            other => panic!("Expected Http, got {other:?}"),
755        }
756    }
757
758    #[test]
759    fn test_humanize_tool_name() {
760        assert_eq!(humanize_tool_name("coding__read_file"), "Read file");
761        assert_eq!(humanize_tool_name("read_file"), "Read file");
762        assert_eq!(humanize_tool_name("bash"), "Bash");
763        assert_eq!(humanize_tool_name("plugins__coding__read_file"), "Read file");
764    }
765
766    #[test]
767    fn test_result_with_result_meta_sets_meta() {
768        use mcp_utils::display_meta::ToolDisplayMeta;
769
770        let session_id = acp::SessionId::new("test-session");
771        let result = ToolCallResult {
772            id: "call_1".to_string(),
773            name: "coding__read_file".to_string(),
774            arguments: "{}".to_string(),
775            result: "file contents".to_string(),
776        };
777        let rm: ToolResultMeta = ToolDisplayMeta::new("Read file", "Cargo.toml, 156 lines").into();
778
779        let notification = map_tool_result_to_notification(session_id, &result, Some(&rm));
780        match notification.update {
781            acp::SessionUpdate::ToolCallUpdate(update) => {
782                assert_eq!(update.fields.title.as_deref(), Some("Read file"), "native title should be set");
783                let meta = update.meta.expect("meta should be present");
784                assert_eq!(
785                    meta.get("display_value").and_then(|v| v.as_str()),
786                    Some("Cargo.toml, 156 lines"),
787                    "display_value should be a flat key in _meta"
788                );
789                assert!(meta.get("display").is_none(), "old nested display object should not be in _meta");
790            }
791            other => panic!("Expected ToolCallUpdate, got {other:?}"),
792        }
793    }
794
795    #[test]
796    fn test_result_without_result_meta() {
797        let session_id = acp::SessionId::new("test-session");
798        let result = ToolCallResult {
799            id: "call_1".to_string(),
800            name: "external__some_tool".to_string(),
801            arguments: "{}".to_string(),
802            result: "ok".to_string(),
803        };
804
805        let notification = map_tool_result_to_notification(session_id, &result, None);
806        match notification.update {
807            acp::SessionUpdate::ToolCallUpdate(update) => {
808                assert!(update.fields.title.is_none());
809                assert!(update.meta.is_none());
810            }
811            other => panic!("Expected ToolCallUpdate, got {other:?}"),
812        }
813    }
814
815    #[test]
816    fn test_plan_notification_extracted_from_result_meta() {
817        use mcp_utils::display_meta::{PlanMeta, PlanMetaEntry, PlanMetaStatus, ToolDisplayMeta};
818
819        let session_id = acp::SessionId::new("test-session");
820        let meta = ToolResultMeta::with_plan(
821            ToolDisplayMeta::new("Todo", "Research AI agents"),
822            PlanMeta {
823                entries: vec![
824                    PlanMetaEntry { content: "Research AI agents".to_string(), status: PlanMetaStatus::InProgress },
825                    PlanMetaEntry { content: "Write tests".to_string(), status: PlanMetaStatus::Pending },
826                ],
827            },
828        );
829
830        let notification = try_extract_plan_notification(session_id, Some(&meta)).expect("should produce plan");
831        match notification.update {
832            acp::SessionUpdate::Plan(plan) => {
833                assert_eq!(plan.entries.len(), 2);
834                assert_eq!(plan.entries[0].content, "Research AI agents");
835                assert_eq!(plan.entries[0].status, acp::PlanEntryStatus::InProgress);
836                assert_eq!(plan.entries[1].content, "Write tests");
837                assert_eq!(plan.entries[1].status, acp::PlanEntryStatus::Pending);
838            }
839            other => panic!("Expected Plan, got {other:?}"),
840        }
841    }
842
843    #[test]
844    fn test_plan_notification_none_when_no_plan_or_no_meta() {
845        use mcp_utils::display_meta::ToolDisplayMeta;
846
847        let sid = acp::SessionId::new("test-session");
848        let meta: ToolResultMeta = ToolDisplayMeta::new("Read file", "main.rs").into();
849        assert!(try_extract_plan_notification(sid.clone(), Some(&meta)).is_none());
850        assert!(try_extract_plan_notification(sid, None).is_none());
851    }
852
853    #[test]
854    fn test_tool_progress_with_display_meta_emits_meta_update() {
855        use mcp_utils::display_meta::ToolDisplayMeta;
856
857        let session_id = acp::SessionId::new("test-session");
858        let meta = ToolResultMeta::from(ToolDisplayMeta::new("Read file", "main.rs"));
859        let serialized = serde_json::to_string(&meta).unwrap();
860
861        let request = ToolCallRequest {
862            id: "call_789".to_string(),
863            name: "coding__read_file".to_string(),
864            arguments: "{}".to_string(),
865        };
866
867        let notification = map_tool_progress_to_notification(session_id, &request, 0.0, None, Some(&serialized))
868            .expect("should produce notification");
869
870        match notification.update {
871            acp::SessionUpdate::ToolCallUpdate(update) => {
872                assert_eq!(&*update.tool_call_id.0, "call_789");
873                assert_eq!(update.fields.title.as_deref(), Some("Read file"), "native title should be set");
874                let meta_map = update.meta.expect("meta should be present");
875                assert_eq!(
876                    meta_map.get("display_value").and_then(|v| v.as_str()),
877                    Some("main.rs"),
878                    "display_value should be a flat key in _meta"
879                );
880                assert!(meta_map.get("display").is_none(), "old nested display object should not be in _meta");
881                assert_eq!(update.fields.status, Some(acp::ToolCallStatus::InProgress));
882                // Should NOT have content (no text progress fallback)
883                assert!(update.fields.content.is_none());
884            }
885            other => panic!("Expected ToolCallUpdate, got {other:?}"),
886        }
887    }
888}
aether_cli/acp/mappers.rs

aether_cli/acp/
mappers.rs
