use aether_core::core::{AgentDeps, Prompt};
use aether_core::events::{
AgentEvent, Command, ContextEvent, MessageEvent, ModelEvent, ToolEvent, TurnEvent, TurnOutcome,
};
use aether_core::mcp::McpHandle;
use aether_telemetry::TelemetryRuntime;
use std::io;
use std::process::ExitCode;
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::error;
use crate::telemetry::build_telemetry_runtime;
use super::error::CliError;
use super::{CliEventKind, RunConfig};
use crate::output::{OutputFormat, print_message};
use crate::runtime::RuntimeBuilder;
use crate::slash_commands::{expand_slash_command, parse_slash_command};
pub async fn run(config: RunConfig) -> Result<ExitCode, CliError> {
setup_tracing(config.verbose);
let telemetry = build_telemetry_runtime(config.telemetry.as_ref(), config.trace_context.clone())?;
let result = run_agent(config, telemetry.clone()).await;
if let Some(telemetry) = telemetry {
telemetry.shutdown_or_log();
}
result
}
async fn run_agent(config: RunConfig, telemetry: Option<Arc<TelemetryRuntime>>) -> Result<ExitCode, CliError> {
let mut spec = config.spec;
if let Some(system_prompt) = config.system_prompt {
spec.prompts.push(Prompt::text(&system_prompt));
}
let registry = config.agent_catalog.registry().clone();
let deps =
AgentDeps::new(config.oauth_credential_store, telemetry.as_ref().map(|runtime| runtime.observer_factory()))
.with_agent_registry(registry);
let (agent, _mcp_snapshot) = RuntimeBuilder::from_spec(config.cwd.clone(), spec)
.mcp_sources(config.mcp_config_sources)
.agent_deps(deps)
.build_ready(vec![])
.await?;
let prompt = expand_prompt(agent.mcp_runtime.handle(), config.prompt).await;
agent
.agent_tx
.send(Command::text(&prompt))
.await
.map_err(|e| CliError::AgentError(format!("Failed to send prompt: {e}")))?;
let exit_code = stream_output(agent.agent_rx, config.output, &config.events).await;
drop(agent.agent_tx);
agent.agent_handle.await_completion().await;
Ok(exit_code)
}
async fn expand_prompt(mcp: &McpHandle, prompt: String) -> String {
let Some(slash_command) = parse_slash_command(&prompt) else {
return prompt;
};
match expand_slash_command(mcp, slash_command.command_name, slash_command.args_text).await {
Ok(expanded) => expanded,
Err(error) => {
error!("Failed to expand slash command: {error}");
prompt
}
}
}
async fn stream_output(mut rx: mpsc::Receiver<AgentEvent>, format: OutputFormat, events: &[CliEventKind]) -> ExitCode {
while let Some(msg) = rx.recv().await {
if should_emit(&msg, events)
&& let Err(error) = print_message(format, &msg)
{
eprintln!("Failed to serialize headless event: {error}");
return ExitCode::FAILURE;
}
if let Some(outcome) = msg.turn_outcome() {
return match outcome {
TurnOutcome::Failed { .. } => ExitCode::FAILURE,
TurnOutcome::Completed | TurnOutcome::Cancelled => ExitCode::SUCCESS,
};
}
}
ExitCode::SUCCESS
}
fn should_emit(msg: &AgentEvent, include: &[CliEventKind]) -> bool {
let Some(kind) = event_kind(msg) else { return false };
include.is_empty() || include.contains(&kind)
}
fn event_kind(msg: &AgentEvent) -> Option<CliEventKind> {
match msg {
AgentEvent::Message(MessageEvent::Text { is_complete: true, .. }) => Some(CliEventKind::Text),
AgentEvent::Message(MessageEvent::Thought { is_complete: true, .. }) => Some(CliEventKind::Thought),
AgentEvent::Tool(ToolEvent::Call { .. }) => Some(CliEventKind::ToolCall),
AgentEvent::Tool(
ToolEvent::Result { .. } | ToolEvent::TaskCreated { .. } | ToolEvent::TaskCompleted { .. },
) => Some(CliEventKind::ToolResult),
AgentEvent::Tool(ToolEvent::Error { .. } | ToolEvent::TaskFailed { .. } | ToolEvent::TaskCancelled { .. }) => {
Some(CliEventKind::ToolError)
}
AgentEvent::Turn(TurnEvent::AutoContinue { .. }) => Some(CliEventKind::AutoContinue),
AgentEvent::Model(ModelEvent::Switched { .. }) => Some(CliEventKind::ModelSwitched),
AgentEvent::Tool(
ToolEvent::Progress { .. }
| ToolEvent::DisplayUpdate { .. }
| ToolEvent::SubAgentProgress { .. }
| ToolEvent::TaskStatus { .. },
) => Some(CliEventKind::ToolProgress),
AgentEvent::Context(ContextEvent::CompactionStarted { .. }) => Some(CliEventKind::ContextCompactionStarted),
AgentEvent::Context(ContextEvent::CompactionEnded { .. }) => Some(CliEventKind::ContextCompactionEnded),
AgentEvent::Context(ContextEvent::CompactionResult { .. }) => Some(CliEventKind::ContextCompactionResult),
AgentEvent::Context(ContextEvent::UsageUpdated { .. }) => Some(CliEventKind::ContextUsage),
AgentEvent::SessionUsage(_) => Some(CliEventKind::SessionUsage),
AgentEvent::Context(ContextEvent::Cleared) => Some(CliEventKind::ContextCleared),
AgentEvent::Turn(TurnEvent::Started { .. }) => Some(CliEventKind::TurnStarted),
AgentEvent::Turn(TurnEvent::Ended { .. }) => Some(CliEventKind::TurnEnded),
AgentEvent::Turn(TurnEvent::RetryScheduled { .. }) => Some(CliEventKind::LlmRetryScheduled),
AgentEvent::Turn(TurnEvent::LlmCallStarted { .. }) => Some(CliEventKind::LlmCallStarted),
AgentEvent::Turn(TurnEvent::LlmCallEnded { .. }) => Some(CliEventKind::LlmCallEnded),
AgentEvent::Tool(ToolEvent::ExecutionStarted { .. }) => Some(CliEventKind::ToolExecutionStarted),
AgentEvent::Tool(ToolEvent::DefinitionsUpdated { .. }) => Some(CliEventKind::ToolDefinitionsUpdated),
AgentEvent::Message(
MessageEvent::Text { is_complete: false, .. } | MessageEvent::Thought { is_complete: false, .. },
)
| AgentEvent::Tool(ToolEvent::CallUpdate { .. }) => None,
}
}
fn setup_tracing(verbose: bool) {
use tracing_subscriber::Layer;
use tracing_subscriber::filter::EnvFilter;
use tracing_subscriber::fmt;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
let filter = if verbose { EnvFilter::new("debug,agent=off") } else { EnvFilter::new("error,agent=off") };
let layer = fmt::layer().with_writer(io::stderr).with_filter(filter);
tracing_subscriber::registry().with(layer).init();
}
#[cfg(test)]
mod tests {
use aether_core::events::StreamState;
use super::*;
use llm::ContextUsage;
#[test]
fn event_kind_none_for_non_output_fragments() {
assert_eq!(event_kind(&AgentEvent::text("id", "x", StreamState::Partial)), None);
assert_eq!(event_kind(&AgentEvent::thought("id", "x", StreamState::Partial)), None);
assert_eq!(
event_kind(&AgentEvent::Tool(ToolEvent::CallUpdate {
tool_call_id: "tc1".to_string(),
chunk: "x".to_string(),
})),
None,
);
}
#[test]
fn event_kind_turn_ended_is_filterable() {
assert_eq!(event_kind(&AgentEvent::turn_ended(TurnOutcome::Completed)), Some(CliEventKind::TurnEnded));
}
#[test]
fn should_emit_empty_filter_rejects_non_output_events() {
assert!(should_emit(&tool_call_msg(), &[]));
assert!(should_emit(
&AgentEvent::Turn(TurnEvent::Ended { outcome: TurnOutcome::Failed { error: "e".to_string() } }),
&[]
));
assert!(should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), &[]));
assert!(!should_emit(&AgentEvent::text("id", "x", StreamState::Partial), &[]));
assert!(!should_emit(
&AgentEvent::Tool(ToolEvent::CallUpdate { tool_call_id: "tc1".to_string(), chunk: "x".to_string() }),
&[],
));
}
#[test]
fn should_emit_single_type_whitelist() {
let filter = &[CliEventKind::ToolCall];
assert!(should_emit(&tool_call_msg(), filter));
assert!(!should_emit(&tool_result_msg(), filter));
assert!(!should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), filter));
}
#[test]
fn should_emit_multi_type_whitelist() {
let filter = &[CliEventKind::ToolCall, CliEventKind::ToolResult];
assert!(should_emit(&tool_call_msg(), filter));
assert!(should_emit(&tool_result_msg(), filter));
assert!(!should_emit(&AgentEvent::turn_ended(TurnOutcome::Completed), filter));
}
#[test]
fn should_emit_turn_ended_respects_filter() {
let msg = AgentEvent::turn_ended(TurnOutcome::Completed);
assert!(should_emit(&msg, &[CliEventKind::TurnEnded]));
assert!(!should_emit(&msg, &[CliEventKind::ToolCall]));
}
#[test]
fn event_kind_covers_every_cli_event_kind() {
use clap::ValueEnum;
let samples = vec![
(AgentEvent::text("id", "x", StreamState::Complete), CliEventKind::Text),
(AgentEvent::thought("id", "x", StreamState::Complete), CliEventKind::Thought),
(tool_call_msg(), CliEventKind::ToolCall),
(tool_result_msg(), CliEventKind::ToolResult),
(
AgentEvent::Tool(ToolEvent::Error {
error: llm::ToolCallError {
id: "tc1".to_string(),
name: "bash".to_string(),
arguments: None,
error: "boom".to_string(),
},
}),
CliEventKind::ToolError,
),
(
AgentEvent::Turn(TurnEvent::AutoContinue {
attempt: 1,
max_attempts: 3,
message_id: llm::MessageId::new(),
content: vec![],
}),
CliEventKind::AutoContinue,
),
(
AgentEvent::Model(ModelEvent::Switched { previous: "a".to_string(), new: "b".to_string() }),
CliEventKind::ModelSwitched,
),
(tool_progress(1.0, None, None), CliEventKind::ToolProgress),
(
AgentEvent::Context(ContextEvent::CompactionStarted {
compaction_id: "compaction".into(),
message_count: 1,
}),
CliEventKind::ContextCompactionStarted,
),
(
AgentEvent::Context(ContextEvent::CompactionEnded {
compaction_id: "compaction".into(),
outcome: aether_core::events::CompactionOutcome::Completed,
}),
CliEventKind::ContextCompactionEnded,
),
(
AgentEvent::Context(ContextEvent::CompactionResult {
compaction_id: "compaction".into(),
message_id: llm::MessageId::new(),
summary: "s".to_string(),
messages_removed: 1,
}),
CliEventKind::ContextCompactionResult,
),
(usage_update(), CliEventKind::ContextUsage),
(
AgentEvent::SessionUsage(llm::testing::session_usage_event(1, llm::TokenUsage::new(1, 1))),
CliEventKind::SessionUsage,
),
(AgentEvent::Context(ContextEvent::Cleared), CliEventKind::ContextCleared),
(AgentEvent::Turn(TurnEvent::Started { content: vec![] }), CliEventKind::TurnStarted),
(AgentEvent::turn_ended(TurnOutcome::Completed), CliEventKind::TurnEnded),
(retry_scheduled(1, 10), CliEventKind::LlmRetryScheduled),
(llm_call_started(0), CliEventKind::LlmCallStarted),
(
AgentEvent::Turn(TurnEvent::LlmCallEnded {
purpose: llm::LlmCallPurpose::Chat,
outcome: aether_core::events::LlmCallOutcome::Cancelled,
}),
CliEventKind::LlmCallEnded,
),
(
AgentEvent::Tool(ToolEvent::ExecutionStarted {
tool_id: "tc1".to_string(),
tool_name: "bash".to_string(),
}),
CliEventKind::ToolExecutionStarted,
),
(AgentEvent::Tool(ToolEvent::DefinitionsUpdated { tools: vec![] }), CliEventKind::ToolDefinitionsUpdated),
];
for kind in CliEventKind::value_variants() {
assert!(samples.iter().any(|(_, k)| k == kind), "samples is missing a case for {kind:?}");
}
for (msg, kind) in &samples {
assert_eq!(event_kind(msg), Some(*kind), "event_kind disagrees for {kind:?}");
}
}
#[tokio::test]
async fn stream_output_turn_ended_breaks_loop_under_filter() {
let (tx, rx) = mpsc::channel(4);
tx.send(AgentEvent::turn_ended(TurnOutcome::Completed)).await.unwrap();
let filter = vec![CliEventKind::ToolCall];
let code = stream_output(rx, OutputFormat::Text, &filter).await;
assert_eq!(code, ExitCode::SUCCESS);
}
#[tokio::test]
async fn stream_output_failed_turn_exits_with_failure() {
let (tx, rx) = mpsc::channel(4);
tx.send(AgentEvent::turn_ended(TurnOutcome::Failed { error: "boom".to_string() })).await.unwrap();
let code = stream_output(rx, OutputFormat::Text, &[]).await;
assert_eq!(code, ExitCode::FAILURE);
}
fn tool_call_msg() -> AgentEvent {
AgentEvent::Tool(ToolEvent::Call {
request: llm::ToolCallRequest {
id: "tc1".to_string(),
name: "bash".to_string(),
arguments: "{}".to_string(),
},
})
}
fn tool_result_msg() -> AgentEvent {
AgentEvent::Tool(ToolEvent::Result {
result: llm::ToolCallResult {
id: "tc1".to_string(),
name: "bash".to_string(),
arguments: "{}".to_string(),
result: "ok".to_string(),
},
result_meta: None,
})
}
fn tool_progress(progress: f64, total: Option<f64>, message: Option<&str>) -> AgentEvent {
AgentEvent::Tool(ToolEvent::Progress {
request: llm::ToolCallRequest {
id: "tc1".to_string(),
name: "bash".to_string(),
arguments: "{}".to_string(),
},
progress,
total,
message: message.map(str::to_string),
})
}
fn retry_scheduled(attempt: u32, delay_ms: u64) -> AgentEvent {
AgentEvent::Turn(TurnEvent::RetryScheduled {
purpose: llm::LlmCallPurpose::Chat,
attempt,
max_attempts: 3,
delay_ms,
})
}
fn llm_call_started(attempt: u32) -> AgentEvent {
AgentEvent::Turn(TurnEvent::LlmCallStarted {
purpose: llm::LlmCallPurpose::Chat,
model: llm::ModelIdentity::default(),
display_name: "test".to_string(),
attempt,
max_attempts: 3,
})
}
fn usage_update() -> AgentEvent {
AgentEvent::Context(ContextEvent::UsageUpdated {
usage: ContextUsage {
input_tokens: 100_000.into(),
context_limit: Some(200_000.into()),
usage_ratio: Some(0.5),
},
})
}
}