1use aether_core::core::{AgentDeps, Prompt};
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 while let Some(msg) = rx.recv().await {
81 if should_emit(&msg, events)
82 && let Err(error) = print_message(format, &msg)
83 {
84 eprintln!("Failed to serialize headless event: {error}");
85 return ExitCode::FAILURE;
86 }
87
88 if let Some(outcome) = msg.turn_outcome() {
89 return match outcome {
90 TurnOutcome::Failed { .. } => ExitCode::FAILURE,
91 TurnOutcome::Completed | TurnOutcome::Cancelled => ExitCode::SUCCESS,
92 };
93 }
94 }
95 ExitCode::SUCCESS
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::ExecutionStarted { .. }) => Some(CliEventKind::ToolExecutionStarted),
134 AgentEvent::Tool(ToolEvent::DefinitionsUpdated { .. }) => Some(CliEventKind::ToolDefinitionsUpdated),
135 AgentEvent::Message(
136 MessageEvent::Text { is_complete: false, .. } | MessageEvent::Thought { is_complete: false, .. },
137 )
138 | AgentEvent::Tool(ToolEvent::CallUpdate { .. }) => 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::CallUpdate {
168 tool_call_id: "tc1".to_string(),
169 chunk: "x".to_string(),
170 })),
171 None,
172 );
173 }
174
175 #[test]
176 fn event_kind_turn_ended_is_filterable() {
177 assert_eq!(event_kind(&AgentEvent::turn_ended(TurnOutcome::Completed)), Some(CliEventKind::TurnEnded));
178 }
179
180 #[test]
181 fn should_emit_empty_filter_rejects_non_output_events() {
182 assert!(should_emit(&tool_call_msg(), &[]));
183 assert!(should_emit(
184 &AgentEvent::Turn(TurnEvent::Ended { outcome: TurnOutcome::Failed { error: "e".to_string() } }),
185 &[]
186 ));
187 assert!(should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), &[]));
188 assert!(!should_emit(&AgentEvent::text("id", "x", StreamState::Partial), &[]));
189 assert!(!should_emit(
190 &AgentEvent::Tool(ToolEvent::CallUpdate { tool_call_id: "tc1".to_string(), chunk: "x".to_string() }),
191 &[],
192 ));
193 }
194
195 #[test]
196 fn should_emit_single_type_whitelist() {
197 let filter = &[CliEventKind::ToolCall];
198 assert!(should_emit(&tool_call_msg(), filter));
199 assert!(!should_emit(&tool_result_msg(), filter));
200 assert!(!should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), filter));
201 }
202
203 #[test]
204 fn should_emit_multi_type_whitelist() {
205 let filter = &[CliEventKind::ToolCall, CliEventKind::ToolResult];
206 assert!(should_emit(&tool_call_msg(), filter));
207 assert!(should_emit(&tool_result_msg(), filter));
208 assert!(!should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), filter));
209 }
210
211 #[test]
212 fn should_emit_turn_ended_respects_filter() {
213 let msg = AgentEvent::turn_ended(TurnOutcome::Completed);
214 assert!(should_emit(&msg, &[CliEventKind::TurnEnded]));
215 assert!(!should_emit(&msg, &[CliEventKind::ToolCall]));
216 }
217
218 #[test]
219 fn event_kind_covers_every_cli_event_kind() {
220 use clap::ValueEnum;
221
222 let samples = vec![
223 (AgentEvent::text("id", "x", StreamState::Complete), CliEventKind::Text),
224 (AgentEvent::thought("id", "x", StreamState::Complete), CliEventKind::Thought),
225 (tool_call_msg(), CliEventKind::ToolCall),
226 (tool_result_msg(), CliEventKind::ToolResult),
227 (
228 AgentEvent::Tool(ToolEvent::Error {
229 error: llm::ToolCallError {
230 id: "tc1".to_string(),
231 name: "bash".to_string(),
232 arguments: None,
233 error: "boom".to_string(),
234 },
235 }),
236 CliEventKind::ToolError,
237 ),
238 (
239 AgentEvent::Turn(TurnEvent::AutoContinue {
240 attempt: 1,
241 max_attempts: 3,
242 message_id: llm::MessageId::new(),
243 content: vec![],
244 }),
245 CliEventKind::AutoContinue,
246 ),
247 (
248 AgentEvent::Model(ModelEvent::Switched { previous: "a".to_string(), new: "b".to_string() }),
249 CliEventKind::ModelSwitched,
250 ),
251 (tool_progress(1.0, None, None), CliEventKind::ToolProgress),
252 (
253 AgentEvent::Context(ContextEvent::CompactionStarted {
254 compaction_id: "compaction".into(),
255 message_count: 1,
256 }),
257 CliEventKind::ContextCompactionStarted,
258 ),
259 (
260 AgentEvent::Context(ContextEvent::CompactionEnded {
261 compaction_id: "compaction".into(),
262 outcome: aether_core::events::CompactionOutcome::Completed,
263 }),
264 CliEventKind::ContextCompactionEnded,
265 ),
266 (
267 AgentEvent::Context(ContextEvent::CompactionResult {
268 compaction_id: "compaction".into(),
269 message_id: llm::MessageId::new(),
270 summary: "s".to_string(),
271 messages_removed: 1,
272 }),
273 CliEventKind::ContextCompactionResult,
274 ),
275 (usage_update(), CliEventKind::ContextUsage),
276 (
277 AgentEvent::SessionUsage(llm::testing::session_usage_event(1, llm::TokenUsage::new(1, 1))),
278 CliEventKind::SessionUsage,
279 ),
280 (AgentEvent::Context(ContextEvent::Cleared), CliEventKind::ContextCleared),
281 (AgentEvent::Turn(TurnEvent::Started { content: vec![] }), CliEventKind::TurnStarted),
282 (AgentEvent::turn_ended(TurnOutcome::Completed), CliEventKind::TurnEnded),
283 (retry_scheduled(1, 10), CliEventKind::LlmRetryScheduled),
284 (llm_call_started(0), CliEventKind::LlmCallStarted),
285 (
286 AgentEvent::Turn(TurnEvent::LlmCallEnded {
287 purpose: llm::LlmCallPurpose::Chat,
288 outcome: aether_core::events::LlmCallOutcome::Cancelled,
289 }),
290 CliEventKind::LlmCallEnded,
291 ),
292 (
293 AgentEvent::Tool(ToolEvent::ExecutionStarted {
294 tool_id: "tc1".to_string(),
295 tool_name: "bash".to_string(),
296 }),
297 CliEventKind::ToolExecutionStarted,
298 ),
299 (AgentEvent::Tool(ToolEvent::DefinitionsUpdated { tools: vec![] }), CliEventKind::ToolDefinitionsUpdated),
300 ];
301
302 for kind in CliEventKind::value_variants() {
303 assert!(samples.iter().any(|(_, k)| k == kind), "samples is missing a case for {kind:?}");
304 }
305
306 for (msg, kind) in &samples {
307 assert_eq!(event_kind(msg), Some(*kind), "event_kind disagrees for {kind:?}");
308 }
309 }
310
311 #[tokio::test]
312 async fn stream_output_turn_ended_breaks_loop_under_filter() {
313 let (tx, rx) = mpsc::channel(4);
314 tx.send(AgentEvent::turn_ended(TurnOutcome::Completed)).await.unwrap();
315 let filter = vec![CliEventKind::ToolCall];
316 let code = stream_output(rx, OutputFormat::Text, &filter).await;
317 assert_eq!(code, ExitCode::SUCCESS);
318 }
319
320 #[tokio::test]
321 async fn stream_output_failed_turn_exits_with_failure() {
322 let (tx, rx) = mpsc::channel(4);
323 tx.send(AgentEvent::turn_ended(TurnOutcome::Failed { error: "boom".to_string() })).await.unwrap();
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}