1use aether_core::core::{AgentDeps, Prompt, recv_agent_event};
2use aether_core::events::{
3 AgentEvent, Command, ContextEvent, MessageEvent, ModelEvent, ToolEvent, TurnEvent, TurnOutcome,
4};
5use aether_core::mcp::McpHandle;
6use aether_telemetry::TelemetryRuntime;
7use std::io;
8use std::process::ExitCode;
9use std::sync::Arc;
10use tokio::sync::mpsc;
11use tracing::error;
12
13use crate::telemetry::build_telemetry_runtime;
14
15use super::error::CliError;
16use super::{CliEventKind, RunConfig};
17use crate::output::{OutputFormat, print_message};
18use crate::runtime::RuntimeBuilder;
19use crate::slash_commands::{expand_slash_command, parse_slash_command};
20
21pub async fn run(config: RunConfig) -> Result<ExitCode, CliError> {
22 setup_tracing(config.verbose);
23
24 let telemetry = build_telemetry_runtime(config.telemetry.as_ref(), config.trace_context.clone())?;
25 let result = run_agent(config, telemetry.clone()).await;
26
27 if let Some(telemetry) = telemetry {
28 telemetry.shutdown_or_log();
29 }
30 result
31}
32
33async fn run_agent(config: RunConfig, telemetry: Option<Arc<TelemetryRuntime>>) -> Result<ExitCode, CliError> {
34 let mut spec = config.spec;
35 if let Some(system_prompt) = config.system_prompt {
36 spec.prompts.push(Prompt::text(&system_prompt));
37 }
38
39 let registry = config.agent_catalog.registry().clone();
40 let deps =
41 AgentDeps::new(config.oauth_credential_store, telemetry.as_ref().map(|runtime| runtime.observer_factory()))
42 .with_agent_registry(registry);
43 let (agent, _mcp_snapshot) = RuntimeBuilder::from_spec(config.cwd.clone(), spec)
44 .mcp_sources(config.mcp_config_sources)
45 .agent_deps(deps)
46 .build_ready(vec![])
47 .await?;
48
49 let prompt = expand_prompt(agent.mcp_runtime.handle(), config.prompt).await;
50
51 agent
52 .agent_tx
53 .send(Command::text(&prompt))
54 .await
55 .map_err(|e| CliError::AgentError(format!("Failed to send prompt: {e}")))?;
56
57 let exit_code = stream_output(agent.agent_rx, config.output, &config.events).await;
58
59 drop(agent.agent_tx);
60 agent.agent_handle.await_completion().await;
61
62 Ok(exit_code)
63}
64
65async fn expand_prompt(mcp: &McpHandle, prompt: String) -> String {
66 let Some(slash_command) = parse_slash_command(&prompt) else {
67 return prompt;
68 };
69
70 match expand_slash_command(mcp, slash_command.command_name, slash_command.args_text).await {
71 Ok(expanded) => expanded,
72 Err(error) => {
73 error!("Failed to expand slash command: {error}");
74 prompt
75 }
76 }
77}
78
79async fn stream_output(mut rx: mpsc::Receiver<AgentEvent>, format: OutputFormat, events: &[CliEventKind]) -> ExitCode {
80 loop {
81 let msg = recv_agent_event(&mut rx).await;
82 if should_emit(&msg, events)
83 && let Err(error) = print_message(format, &msg)
84 {
85 eprintln!("Failed to serialize headless event: {error}");
86 return ExitCode::FAILURE;
87 }
88
89 if let Some(outcome) = msg.turn_outcome() {
90 return match outcome {
91 TurnOutcome::Failed { .. } => ExitCode::FAILURE,
92 TurnOutcome::Completed | TurnOutcome::Cancelled => ExitCode::SUCCESS,
93 };
94 }
95 }
96}
97
98fn should_emit(msg: &AgentEvent, include: &[CliEventKind]) -> bool {
99 let Some(kind) = event_kind(msg) else { return false };
100 include.is_empty() || include.contains(&kind)
101}
102
103fn event_kind(msg: &AgentEvent) -> Option<CliEventKind> {
104 match msg {
105 AgentEvent::Message(MessageEvent::Text { is_complete: true, .. }) => Some(CliEventKind::Text),
106 AgentEvent::Message(MessageEvent::Thought { is_complete: true, .. }) => Some(CliEventKind::Thought),
107 AgentEvent::Tool(ToolEvent::Call { .. }) => Some(CliEventKind::ToolCall),
108 AgentEvent::Tool(
109 ToolEvent::Result { .. } | ToolEvent::TaskCreated { .. } | ToolEvent::TaskCompleted { .. },
110 ) => Some(CliEventKind::ToolResult),
111 AgentEvent::Tool(ToolEvent::Error { .. } | ToolEvent::TaskFailed { .. } | ToolEvent::TaskCancelled { .. }) => {
112 Some(CliEventKind::ToolError)
113 }
114 AgentEvent::Turn(TurnEvent::AutoContinue { .. }) => Some(CliEventKind::AutoContinue),
115 AgentEvent::Model(ModelEvent::Switched { .. }) => Some(CliEventKind::ModelSwitched),
116 AgentEvent::Tool(
117 ToolEvent::Progress { .. }
118 | ToolEvent::DisplayUpdate { .. }
119 | ToolEvent::SubAgentProgress { .. }
120 | ToolEvent::TaskStatus { .. },
121 ) => Some(CliEventKind::ToolProgress),
122 AgentEvent::Context(ContextEvent::CompactionStarted { .. }) => Some(CliEventKind::ContextCompactionStarted),
123 AgentEvent::Context(ContextEvent::CompactionEnded { .. }) => Some(CliEventKind::ContextCompactionEnded),
124 AgentEvent::Context(ContextEvent::CompactionResult { .. }) => Some(CliEventKind::ContextCompactionResult),
125 AgentEvent::Context(ContextEvent::UsageUpdated { .. }) => Some(CliEventKind::ContextUsage),
126 AgentEvent::SessionUsage(_) => Some(CliEventKind::SessionUsage),
127 AgentEvent::Context(ContextEvent::Cleared) => Some(CliEventKind::ContextCleared),
128 AgentEvent::Turn(TurnEvent::Started { .. }) => Some(CliEventKind::TurnStarted),
129 AgentEvent::Turn(TurnEvent::Ended { .. }) => Some(CliEventKind::TurnEnded),
130 AgentEvent::Turn(TurnEvent::RetryScheduled { .. }) => Some(CliEventKind::LlmRetryScheduled),
131 AgentEvent::Turn(TurnEvent::LlmCallStarted { .. }) => Some(CliEventKind::LlmCallStarted),
132 AgentEvent::Turn(TurnEvent::LlmCallEnded { .. }) => Some(CliEventKind::LlmCallEnded),
133 AgentEvent::Tool(ToolEvent::DefinitionsUpdated { .. }) => Some(CliEventKind::ToolDefinitionsUpdated),
134 AgentEvent::Message(
135 MessageEvent::Text { is_complete: false, .. } | MessageEvent::Thought { is_complete: false, .. },
136 )
137 | AgentEvent::Turn(TurnEvent::UserMessageInserted { .. } | TurnEvent::UserMessageDiscarded { .. })
138 | AgentEvent::Tool(ToolEvent::InputStarted { .. } | ToolEvent::InputDelta { .. }) => None,
139 }
140}
141
142fn setup_tracing(verbose: bool) {
143 use tracing_subscriber::Layer;
144 use tracing_subscriber::filter::EnvFilter;
145 use tracing_subscriber::fmt;
146 use tracing_subscriber::layer::SubscriberExt;
147 use tracing_subscriber::util::SubscriberInitExt;
148
149 let filter = if verbose { EnvFilter::new("debug,agent=off") } else { EnvFilter::new("error,agent=off") };
150 let layer = fmt::layer().with_writer(io::stderr).with_filter(filter);
151
152 tracing_subscriber::registry().with(layer).init();
153}
154
155#[cfg(test)]
156mod tests {
157 use aether_core::events::StreamState;
158
159 use super::*;
160 use llm::ContextUsage;
161
162 #[test]
163 fn event_kind_none_for_non_output_fragments() {
164 assert_eq!(event_kind(&AgentEvent::text("id", "x", StreamState::Partial)), None);
165 assert_eq!(event_kind(&AgentEvent::thought("id", "x", StreamState::Partial)), None);
166 assert_eq!(
167 event_kind(&AgentEvent::Tool(ToolEvent::InputStarted { id: "tc1".to_string(), name: "bash".to_string() })),
168 None,
169 );
170 assert_eq!(
171 event_kind(&AgentEvent::Tool(ToolEvent::InputDelta { id: "tc1".to_string(), chunk: "x".to_string() })),
172 None,
173 );
174 }
175
176 #[test]
177 fn event_kind_turn_ended_is_filterable() {
178 assert_eq!(event_kind(&AgentEvent::turn_ended(TurnOutcome::Completed)), Some(CliEventKind::TurnEnded));
179 }
180
181 #[test]
182 fn should_emit_empty_filter_rejects_non_output_events() {
183 assert!(should_emit(&tool_call_msg(), &[]));
184 assert!(should_emit(&AgentEvent::turn_ended(TurnOutcome::failed("e")), &[]));
185 assert!(should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), &[]));
186 assert!(!should_emit(&AgentEvent::text("id", "x", StreamState::Partial), &[]));
187 assert!(!should_emit(
188 &AgentEvent::Tool(ToolEvent::InputDelta { id: "tc1".to_string(), chunk: "x".to_string() }),
189 &[],
190 ));
191 }
192
193 #[test]
194 fn should_emit_single_type_whitelist() {
195 let filter = &[CliEventKind::ToolCall];
196 assert!(should_emit(&tool_call_msg(), filter));
197 assert!(!should_emit(&tool_result_msg(), filter));
198 assert!(!should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), filter));
199 }
200
201 #[test]
202 fn should_emit_multi_type_whitelist() {
203 let filter = &[CliEventKind::ToolCall, CliEventKind::ToolResult];
204 assert!(should_emit(&tool_call_msg(), filter));
205 assert!(should_emit(&tool_result_msg(), filter));
206 assert!(!should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), filter));
207 }
208
209 #[test]
210 fn should_emit_turn_ended_respects_filter() {
211 let msg = AgentEvent::turn_ended(TurnOutcome::Completed);
212 assert!(should_emit(&msg, &[CliEventKind::TurnEnded]));
213 assert!(!should_emit(&msg, &[CliEventKind::ToolCall]));
214 }
215
216 #[test]
217 fn event_kind_covers_every_cli_event_kind() {
218 use clap::ValueEnum;
219
220 let samples = vec![
221 (AgentEvent::text("id", "x", StreamState::Complete), CliEventKind::Text),
222 (AgentEvent::thought("id", "x", StreamState::Complete), CliEventKind::Thought),
223 (tool_call_msg(), CliEventKind::ToolCall),
224 (tool_result_msg(), CliEventKind::ToolResult),
225 (
226 AgentEvent::Tool(ToolEvent::Error {
227 error: llm::ToolCallError {
228 id: "tc1".to_string(),
229 name: "bash".to_string(),
230 arguments: None,
231 error: "boom".to_string(),
232 },
233 }),
234 CliEventKind::ToolError,
235 ),
236 (
237 AgentEvent::Turn(TurnEvent::AutoContinue {
238 attempt: 1,
239 max_attempts: 3,
240 message_id: llm::MessageId::new(),
241 content: vec![],
242 }),
243 CliEventKind::AutoContinue,
244 ),
245 (
246 AgentEvent::Model(ModelEvent::Switched { previous: "a".to_string(), new: "b".to_string() }),
247 CliEventKind::ModelSwitched,
248 ),
249 (tool_progress(1.0, None, None), CliEventKind::ToolProgress),
250 (
251 AgentEvent::Context(ContextEvent::CompactionStarted {
252 compaction_id: "compaction".into(),
253 message_count: 1,
254 }),
255 CliEventKind::ContextCompactionStarted,
256 ),
257 (
258 AgentEvent::Context(ContextEvent::CompactionEnded {
259 compaction_id: "compaction".into(),
260 outcome: aether_core::events::CompactionOutcome::Completed,
261 }),
262 CliEventKind::ContextCompactionEnded,
263 ),
264 (
265 AgentEvent::Context(ContextEvent::CompactionResult {
266 compaction_id: "compaction".into(),
267 message_id: llm::MessageId::new(),
268 summary: "s".to_string(),
269 messages_removed: 1,
270 }),
271 CliEventKind::ContextCompactionResult,
272 ),
273 (usage_update(), CliEventKind::ContextUsage),
274 (
275 AgentEvent::SessionUsage(llm::testing::session_usage_event(1, llm::TokenUsage::new(1, 1))),
276 CliEventKind::SessionUsage,
277 ),
278 (AgentEvent::Context(ContextEvent::Cleared), CliEventKind::ContextCleared),
279 (AgentEvent::Turn(TurnEvent::Started { content: vec![] }), CliEventKind::TurnStarted),
280 (AgentEvent::turn_ended(TurnOutcome::Completed), CliEventKind::TurnEnded),
281 (retry_scheduled(1, 10), CliEventKind::LlmRetryScheduled),
282 (llm_call_started(0), CliEventKind::LlmCallStarted),
283 (
284 AgentEvent::Turn(TurnEvent::LlmCallEnded {
285 purpose: llm::LlmCallPurpose::Chat,
286 outcome: aether_core::events::LlmCallOutcome::Cancelled,
287 }),
288 CliEventKind::LlmCallEnded,
289 ),
290 (AgentEvent::Tool(ToolEvent::DefinitionsUpdated { tools: vec![] }), CliEventKind::ToolDefinitionsUpdated),
291 ];
292
293 for kind in CliEventKind::value_variants() {
294 assert!(samples.iter().any(|(_, k)| k == kind), "samples is missing a case for {kind:?}");
295 }
296
297 for (msg, kind) in &samples {
298 assert_eq!(event_kind(msg), Some(*kind), "event_kind disagrees for {kind:?}");
299 }
300 }
301
302 #[tokio::test]
303 async fn stream_output_turn_ended_breaks_loop_under_filter() {
304 let (tx, rx) = mpsc::channel(4);
305 tx.send(AgentEvent::turn_ended(TurnOutcome::Completed)).await.unwrap();
306 let filter = vec![CliEventKind::ToolCall];
307 let code = stream_output(rx, OutputFormat::Text, &filter).await;
308 assert_eq!(code, ExitCode::SUCCESS);
309 }
310
311 #[tokio::test]
312 async fn stream_output_failed_turn_exits_with_failure() {
313 let (tx, rx) = mpsc::channel(4);
314 tx.send(AgentEvent::turn_ended(TurnOutcome::failed("boom"))).await.unwrap();
315 let code = stream_output(rx, OutputFormat::Text, &[]).await;
316 assert_eq!(code, ExitCode::FAILURE);
317 }
318
319 #[tokio::test]
320 async fn stream_output_channel_closed_before_turn_ended_exits_with_failure() {
321 let (tx, rx) = mpsc::channel(4);
322 tx.send(AgentEvent::Turn(TurnEvent::Started { content: vec![] })).await.unwrap();
323 drop(tx);
324 let code = stream_output(rx, OutputFormat::Text, &[]).await;
325 assert_eq!(code, ExitCode::FAILURE);
326 }
327
328 fn tool_call_msg() -> AgentEvent {
329 AgentEvent::Tool(ToolEvent::Call {
330 request: llm::ToolCallRequest {
331 id: "tc1".to_string(),
332 name: "bash".to_string(),
333 arguments: "{}".to_string(),
334 },
335 })
336 }
337
338 fn tool_result_msg() -> AgentEvent {
339 AgentEvent::Tool(ToolEvent::Result {
340 result: llm::ToolCallResult {
341 id: "tc1".to_string(),
342 name: "bash".to_string(),
343 arguments: "{}".to_string(),
344 result: "ok".to_string(),
345 },
346 result_meta: None,
347 })
348 }
349
350 fn tool_progress(progress: f64, total: Option<f64>, message: Option<&str>) -> AgentEvent {
351 AgentEvent::Tool(ToolEvent::Progress {
352 request: llm::ToolCallRequest {
353 id: "tc1".to_string(),
354 name: "bash".to_string(),
355 arguments: "{}".to_string(),
356 },
357 progress,
358 total,
359 message: message.map(str::to_string),
360 })
361 }
362
363 fn retry_scheduled(attempt: u32, delay_ms: u64) -> AgentEvent {
364 AgentEvent::Turn(TurnEvent::RetryScheduled {
365 purpose: llm::LlmCallPurpose::Chat,
366 attempt,
367 max_attempts: 3,
368 delay_ms,
369 })
370 }
371
372 fn llm_call_started(attempt: u32) -> AgentEvent {
373 AgentEvent::Turn(TurnEvent::LlmCallStarted {
374 purpose: llm::LlmCallPurpose::Chat,
375 model: llm::ModelIdentity::default(),
376 display_name: "test".to_string(),
377 attempt,
378 max_attempts: 3,
379 })
380 }
381
382 fn usage_update() -> AgentEvent {
383 AgentEvent::Context(ContextEvent::UsageUpdated {
384 usage: ContextUsage {
385 input_tokens: 100_000.into(),
386 context_limit: Some(200_000.into()),
387 usage_ratio: Some(0.5),
388 },
389 })
390 }
391}