use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use nemo_relay::observability::plugin_component::{OBSERVABILITY_PLUGIN_KIND, ObservabilityConfig};
use nemo_relay::plugin::PluginConfig;
use reqwest::Client;
use serde_json::{Value, json};
use tokio::net::TcpListener;
use tokio::process::Command;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use crate::config::{
AgentConfigs, CodingAgent, EasyPathCommand, GatewayConfig, ResolvedConfig, RunCommand,
ServerArgs, any_config_file_exists, resolve_run_config,
};
use crate::error::CliError;
use crate::installer::{generated_hooks, hook_forward_command, merge_hooks, read_json_file};
use crate::server;
pub(crate) async fn run(
command: RunCommand,
inherited: Option<&ServerArgs>,
) -> Result<ExitCode, CliError> {
let run = TransparentRun::new(command, inherited).await?;
run.print_if_requested();
run.execute().await
}
pub(crate) async fn easy_path(
agent: CodingAgent,
command: EasyPathCommand,
inherited: Option<&ServerArgs>,
) -> Result<ExitCode, CliError> {
let explicit_config = inherited.and_then(|args| args.config.as_deref());
let needs_setup = match explicit_config {
Some(path) => !path.exists(),
None => !any_config_file_exists(),
};
if needs_setup {
crate::setup::run(Some(agent)).await?;
}
let synthetic = RunCommand {
agent: Some(agent),
config: explicit_config.map(std::path::Path::to_path_buf),
openai_base_url: None,
anthropic_base_url: None,
session_metadata: None,
plugin_config: None,
dry_run: false,
print: false,
command: command.command,
};
run(synthetic, inherited).await
}
struct TransparentRun {
agent: CodingAgent,
prepared: PreparedRun,
resolved: ResolvedConfig,
listener: TcpListener,
gateway_url: String,
dry_run: bool,
print: bool,
}
impl TransparentRun {
async fn new(command: RunCommand, inherited: Option<&ServerArgs>) -> Result<Self, CliError> {
let dry_run = command.dry_run;
let print = command.print;
let mut resolved = resolve_run_config(&command, inherited)?;
let (agent, argv) = resolve_agent_and_argv(&command, &resolved.agents)?;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let address = listener.local_addr()?;
let gateway_url = format!("http://{address}");
resolved.gateway.bind = address;
let prepared = PreparedRun::new(agent, argv, &gateway_url, &resolved, dry_run)?;
Ok(Self {
agent,
prepared,
resolved,
listener,
gateway_url,
dry_run,
print,
})
}
fn print_if_requested(&self) {
if self.print || self.dry_run {
self.prepared
.print(self.agent, &self.gateway_url, &self.resolved);
}
}
async fn execute(self) -> Result<ExitCode, CliError> {
if self.dry_run {
return Ok(ExitCode::SUCCESS);
}
self.prepared
.print_live_status(self.agent, &self.gateway_url, &self.resolved);
execute_live_run(
self.listener,
self.resolved.gateway,
&self.gateway_url,
self.prepared,
)
.await
}
}
async fn execute_live_run(
listener: TcpListener,
gateway_config: GatewayConfig,
gateway_url: &str,
prepared: PreparedRun,
) -> Result<ExitCode, CliError> {
let running_server = RunningGateway::start(listener, gateway_config);
if let Err(error) = wait_for_health(gateway_url).await {
let _ = running_server.stop().await;
return Err(error);
}
let status = prepared.spawn_and_wait().await;
let restore = prepared.restore();
let server_result = running_server.stop().await;
restore?;
server_result?;
Ok(exit_code(status?))
}
fn resolve_agent_and_argv(
command: &RunCommand,
agents: &AgentConfigs,
) -> Result<(CodingAgent, Vec<String>), CliError> {
let argv = resolved_argv(command, agents)?;
let agent = resolved_agent(command, &argv)?;
Ok((agent, argv))
}
fn resolved_argv(command: &RunCommand, agents: &AgentConfigs) -> Result<Vec<String>, CliError> {
if let Some(agent) = command.agent {
let mut argv = configured_command(agent, agents)
.unwrap_or_else(|| vec![default_command_for(agent).to_string()]);
argv.extend(command.command.iter().cloned());
return Ok(argv);
}
if command.command.is_empty() {
return Err(CliError::Launch(
"missing command; pass -- <agent-command> or --agent with a configured command".into(),
));
}
Ok(command.command.clone())
}
const fn default_command_for(agent: CodingAgent) -> &'static str {
match agent {
CodingAgent::ClaudeCode => "claude",
CodingAgent::Codex => "codex",
CodingAgent::Cursor => "cursor-agent",
CodingAgent::Hermes => "hermes",
}
}
fn resolved_agent(command: &RunCommand, argv: &[String]) -> Result<CodingAgent, CliError> {
if let Some(agent) = command.agent {
return Ok(agent);
}
CodingAgent::infer(&argv[0]).ok_or_else(|| {
CliError::Launch(format!(
"could not infer coding agent from command {:?}; pass --agent claude, --agent codex, --agent cursor, or --agent hermes",
argv[0]
))
})
}
fn configured_command(agent: CodingAgent, agents: &AgentConfigs) -> Option<Vec<String>> {
let command = match agent {
CodingAgent::ClaudeCode => agents.claude.command.as_ref(),
CodingAgent::Codex => agents.codex.command.as_ref(),
CodingAgent::Cursor => agents.cursor.command.as_ref(),
CodingAgent::Hermes => agents.hermes.command.as_ref(),
}?;
let argv: Vec<_> = command.split_whitespace().map(ToOwned::to_owned).collect();
(!argv.is_empty()).then_some(argv)
}
struct PreparedRun {
argv: Vec<String>,
env: Vec<(String, String)>,
temp_dirs: Vec<PathBuf>,
cursor_restore: Option<CursorRestore>,
notes: Vec<String>,
}
struct CursorRestore {
path: PathBuf,
backup_path: Option<PathBuf>,
had_original: bool,
}
struct RunningGateway {
shutdown_tx: oneshot::Sender<()>,
task: JoinHandle<Result<(), CliError>>,
}
impl RunningGateway {
fn start(listener: TcpListener, config: crate::config::GatewayConfig) -> Self {
let (shutdown_tx, shutdown_rx) = oneshot::channel();
let task = tokio::spawn(async move {
server::serve_listener(listener, config, Some(shutdown_rx)).await
});
Self { shutdown_tx, task }
}
async fn stop(self) -> Result<(), CliError> {
let _ = self.shutdown_tx.send(());
self.task
.await
.map_err(|error| CliError::Launch(format!("gateway task failed: {error}")))?
}
}
impl PreparedRun {
fn new(
agent: CodingAgent,
argv: Vec<String>,
gateway_url: &str,
resolved: &ResolvedConfig,
dry_run: bool,
) -> Result<Self, CliError> {
let mut run = Self {
argv,
env: vec![("NEMO_RELAY_GATEWAY_URL".into(), gateway_url.into())],
temp_dirs: Vec::new(),
cursor_restore: None,
notes: Vec::new(),
};
if let Some(path) = path_with_transparent_hook_dir() {
run.env.push(("PATH".into(), path));
}
match agent {
CodingAgent::ClaudeCode => {
if dry_run {
run.prepare_claude_dry(gateway_url);
} else {
run.prepare_claude(gateway_url)?;
}
}
CodingAgent::Codex => run.prepare_codex(gateway_url),
CodingAgent::Cursor => {
if resolved.agents.cursor.patch_restore_hooks {
if dry_run {
run.prepare_cursor_dry()?;
} else {
run.prepare_cursor()?;
}
}
}
CodingAgent::Hermes => run.prepare_hermes(resolved.agents.hermes.hooks_path.as_deref()),
}
Ok(run)
}
fn prepare_claude_dry(&mut self, gateway_url: &str) {
insert_after_agent(
&mut self.argv,
CodingAgent::ClaudeCode,
[
"--plugin-dir".into(),
"<temporary-claude-plugin-dir>".into(),
],
);
self.env
.push(("ANTHROPIC_BASE_URL".into(), gateway_url.to_string()));
self.notes
.push("would generate a temporary Claude Code plugin directory".into());
}
fn prepare_claude(&mut self, gateway_url: &str) -> Result<(), CliError> {
let root = temp_dir("nemo-relay-claude-plugin")?;
std::fs::create_dir_all(root.join(".claude-plugin"))?;
std::fs::create_dir_all(root.join("hooks"))?;
std::fs::write(
root.join(".claude-plugin/plugin.json"),
serde_json::to_vec_pretty(&json!({
"name": "nemo-relay-cli",
"version": env!("CARGO_PKG_VERSION"),
"description": "Temporary NeMo Relay gateway hooks"
}))
.map_err(|error| CliError::Launch(error.to_string()))?,
)?;
write_hooks(
&root.join("hooks/hooks.json"),
generated_hooks(
CodingAgent::ClaudeCode,
&hook_forward_command(&transparent_hook_executable(), CodingAgent::ClaudeCode),
),
)?;
insert_after_agent(
&mut self.argv,
CodingAgent::ClaudeCode,
["--plugin-dir".into(), root.display().to_string()],
);
self.env
.push(("ANTHROPIC_BASE_URL".into(), gateway_url.to_string()));
self.temp_dirs.push(root);
Ok(())
}
fn prepare_codex(&mut self, gateway_url: &str) {
let has_openai_key = std::env::var("OPENAI_API_KEY")
.ok()
.is_some_and(|v| !v.is_empty());
let has_codex_auth = std::env::var_os("HOME")
.or_else(|| std::env::var_os("USERPROFILE"))
.map(|h| {
std::path::PathBuf::from(h)
.join(".codex/auth.json")
.exists()
})
.unwrap_or(false);
if !has_openai_key && !has_codex_auth {
eprintln!(
"warning: No OpenAI credentials found. Either export OPENAI_API_KEY \
(e.g. `export OPENAI_API_KEY=sk-...`), log in to codex (`codex --login`), \
or pass `--openai-base-url` to an upstream that needs no key."
);
}
let hook_command = hook_forward_command(&transparent_hook_executable(), CodingAgent::Codex);
let mut args = vec![
"--config".to_string(),
"features.hooks=true".to_string(),
"--config".to_string(),
"model_provider=\"nemo-relay-openai\"".to_string(),
"--config".to_string(),
codex_gateway_provider_config(gateway_url),
];
for (event, groups) in generated_hooks(CodingAgent::Codex, &hook_command)["hooks"]
.as_object()
.into_iter()
.flatten()
{
args.push("--config".to_string());
args.push(format!("hooks.{event}={}", hook_groups_toml(groups)));
}
insert_after_agent(&mut self.argv, CodingAgent::Codex, args);
}
fn prepare_cursor(&mut self) -> Result<(), CliError> {
let path = cursor_hooks_path()?;
let (had_original, backup_path) = backup_existing_cursor_hooks(&path)?;
write_merged_cursor_hooks(&path)?;
self.cursor_restore = Some(CursorRestore {
path,
backup_path,
had_original,
});
Ok(())
}
fn prepare_cursor_dry(&mut self) -> Result<(), CliError> {
let path = cursor_hooks_path()?;
self.notes.push(format!(
"would temporarily merge NeMo Relay hooks into {}",
path.display()
));
Ok(())
}
fn prepare_hermes(&mut self, hooks_path: Option<&std::path::Path>) {
let note = match hooks_path {
Some(path) => format!(
"Hermes hooks at {} — re-run `nemo-relay config hermes` to refresh.",
path.display()
),
None => "Hermes hooks not yet installed — run `nemo-relay config hermes` once so hermes traces under this gateway.".into(),
};
self.notes.push(note);
}
async fn spawn_and_wait(&self) -> Result<std::process::ExitStatus, CliError> {
let mut command = Command::new(&self.argv[0]);
command.args(&self.argv[1..]);
for (name, value) in &self.env {
command.env(name, value);
}
let mut child = command.spawn()?;
child.wait().await.map_err(CliError::from)
}
fn restore(&self) -> Result<(), CliError> {
for dir in &self.temp_dirs {
let _ = std::fs::remove_dir_all(dir);
}
let Some(cursor) = &self.cursor_restore else {
return Ok(());
};
match (&cursor.backup_path, cursor.had_original) {
(Some(backup), true) => {
std::fs::copy(backup, &cursor.path).map_err(|error| {
CliError::Launch(format!(
"failed to restore Cursor hooks from {}: {error}",
backup.display()
))
})?;
let _ = std::fs::remove_file(backup);
}
(_, false) if cursor.path.exists() => {
std::fs::remove_file(&cursor.path).map_err(|error| {
CliError::Launch(format!(
"failed to remove temporary Cursor hooks {}: {error}",
cursor.path.display()
))
})?;
}
_ => {}
}
Ok(())
}
fn print_live_status(&self, agent: CodingAgent, gateway_url: &str, resolved: &ResolvedConfig) {
if !std::io::IsTerminal::is_terminal(&std::io::stdout()) {
return;
}
let mut lines: Vec<String> = Vec::new();
lines.push(format!("NeMo Relay → {}", agent.as_arg()));
lines.push(format!(" Gateway {gateway_url}"));
let destinations = exporter_destinations(&resolved.gateway);
if destinations.is_empty() {
lines.push(" Exporters not configured".into());
} else {
for (index, destination) in destinations.iter().enumerate() {
lines.push(format!(
" {}{}",
if index == 0 {
"Exporters "
} else {
" "
},
destination
));
}
}
if !self.notes.is_empty() {
lines.push(String::new());
for note in &self.notes {
lines.push(format!("⚠ {note}"));
}
}
let use_color = std::io::IsTerminal::is_terminal(&std::io::stderr())
&& std::env::var_os("NO_COLOR").is_none();
let max_w = lines.iter().map(|l| l.chars().count()).max().unwrap_or(0);
let inner = max_w + 2;
eprintln!();
eprint_border_line('╭', '╮', inner, use_color);
for line in &lines {
let pad = max_w - line.chars().count();
let body = format!(" {line}{spaces} ", spaces = " ".repeat(pad));
if use_color {
eprintln!("\x1b[38;5;112m│\x1b[0m{body}\x1b[38;5;112m│\x1b[0m");
} else {
eprintln!("│{body}│");
}
}
eprint_border_line('╰', '╯', inner, use_color);
eprintln!();
}
fn print(&self, agent: CodingAgent, gateway_url: &str, resolved: &ResolvedConfig) {
println!("agent = {}", agent.as_arg());
println!("gateway_url = {gateway_url}");
println!("openai_base_url = {}", resolved.gateway.openai_base_url);
println!(
"anthropic_base_url = {}",
resolved.gateway.anthropic_base_url
);
let destinations = exporter_destinations(&resolved.gateway);
if destinations.is_empty() {
println!("exporters = not_configured");
} else {
for destination in destinations {
println!("exporter = {destination}");
}
}
println!("argv = {}", self.argv.join(" "));
for (name, value) in &self.env {
println!("env.{name} = {value}");
}
if let Some(cursor) = &self.cursor_restore {
println!("cursor_hooks = {}", cursor.path.display());
}
for note in &self.notes {
println!("note = {note}");
}
}
}
fn exporter_destinations(config: &GatewayConfig) -> Vec<String> {
let Some(plugin_config) = config.plugin_config.as_ref() else {
return Vec::new();
};
let Ok(plugin_config) = serde_json::from_value::<PluginConfig>(plugin_config.clone()) else {
return vec!["configured (invalid plugin config)".into()];
};
let Some(component) = plugin_config
.components
.iter()
.find(|component| component.kind == OBSERVABILITY_PLUGIN_KIND)
else {
return Vec::new();
};
if !component.enabled {
return Vec::new();
}
let Ok(observability) =
serde_json::from_value::<ObservabilityConfig>(Value::Object(component.config.clone()))
else {
return vec!["Observability configured (invalid config)".into()];
};
observability_exporter_destinations(&observability)
}
fn observability_exporter_destinations(config: &ObservabilityConfig) -> Vec<String> {
let mut destinations = Vec::new();
if let Some(section) = config.atof.as_ref().filter(|section| section.enabled) {
let directory = section
.output_directory
.clone()
.unwrap_or_else(current_output_directory);
let path = directory.join(
section
.filename
.clone()
.unwrap_or_else(|| "nemo-relay-events-<timestamp>.jsonl".into()),
);
destinations.push(format!("ATOF {}", path.display()));
}
if let Some(section) = config.atif.as_ref().filter(|section| section.enabled) {
let directory = section
.output_directory
.clone()
.unwrap_or_else(current_output_directory);
destinations.push(format!(
"ATIF {}",
directory.join(§ion.filename_template).display()
));
}
if let Some(section) = config
.opentelemetry
.as_ref()
.filter(|section| section.enabled)
{
destinations.push(format!(
"OpenTelemetry {}",
section
.endpoint
.as_deref()
.unwrap_or("OTLP endpoint from environment/default")
));
}
if let Some(section) = config
.openinference
.as_ref()
.filter(|section| section.enabled)
{
destinations.push(format!(
"OpenInference {}",
section
.endpoint
.as_deref()
.unwrap_or("OTLP endpoint from environment/default")
));
}
destinations
}
fn current_output_directory() -> PathBuf {
std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
}
fn exit_code(status: std::process::ExitStatus) -> ExitCode {
status
.code()
.and_then(|code| u8::try_from(code).ok())
.map(ExitCode::from)
.unwrap_or(ExitCode::FAILURE)
}
async fn wait_for_health(gateway_url: &str) -> Result<(), CliError> {
crate::tls::install_rustls_crypto_provider();
let client = Client::new();
let url = format!("{}/healthz", gateway_url.trim_end_matches('/'));
for _ in 0..50 {
if let Ok(response) = client.get(&url).send().await
&& response.status().is_success()
{
return Ok(());
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
Err(CliError::Launch(format!(
"gateway did not become ready at {url}"
)))
}
fn codex_gateway_provider_config(gateway_url: &str) -> String {
format!(
"model_providers.nemo-relay-openai={{name=\"NeMo Relay OpenAI\",base_url={},wire_api=\"responses\",requires_openai_auth=true,supports_websockets=false}}",
toml_string(gateway_url)
)
}
fn eprint_border_line(left: char, right: char, inner_width: usize, color: bool) {
let dashes = "─".repeat(inner_width);
if color {
eprintln!("\x1b[38;5;112m{left}{dashes}{right}\x1b[0m");
} else {
eprintln!("{left}{dashes}{right}");
}
}
fn transparent_hook_executable() -> String {
std::env::current_exe()
.ok()
.and_then(|path| path.to_str().map(str::to_owned))
.unwrap_or_else(|| "nemo-relay".to_string())
}
fn path_with_transparent_hook_dir() -> Option<String> {
let dir = std::env::current_exe()
.ok()
.and_then(|path| path.parent().map(Path::to_path_buf))?;
let mut paths: Vec<PathBuf> = std::env::var_os("PATH")
.as_deref()
.map(std::env::split_paths)
.into_iter()
.flatten()
.collect();
if !paths.iter().any(|path| path == &dir) {
paths.push(dir);
}
std::env::join_paths(paths)
.ok()
.map(|path| path.to_string_lossy().into_owned())
}
fn insert_after_agent(
argv: &mut Vec<String>,
agent: CodingAgent,
args: impl IntoIterator<Item = String>,
) {
let index = argv
.iter()
.enumerate()
.filter_map(|(index, arg)| (CodingAgent::infer(arg) == Some(agent)).then_some(index))
.next_back()
.unwrap_or(0);
argv.splice(index + 1..index + 1, args);
}
fn write_hooks(path: &Path, hooks: Value) -> Result<(), CliError> {
std::fs::write(
path,
serde_json::to_vec_pretty(&hooks).map_err(|error| CliError::Launch(error.to_string()))?,
)?;
Ok(())
}
fn backup_existing_cursor_hooks(path: &Path) -> Result<(bool, Option<PathBuf>), CliError> {
let had_original = path.exists();
if !had_original {
return Ok((false, None));
}
let backup = path.with_extension(format!("json.nemo-relay-run.bak.{}", timestamp()?));
std::fs::copy(path, &backup)?;
Ok((true, Some(backup)))
}
fn write_merged_cursor_hooks(path: &Path) -> Result<(), CliError> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut merged = merge_hooks(
read_json_file(path)?,
generated_hooks(
CodingAgent::Cursor,
&hook_forward_command(&transparent_hook_executable(), CodingAgent::Cursor),
),
)?;
if let Some(root) = merged.as_object_mut() {
root.insert("version".to_string(), json!(1));
}
let contents = serde_json::to_string_pretty(&merged)
.map_err(|error| CliError::Launch(error.to_string()))?;
std::fs::write(path, contents)?;
Ok(())
}
fn hook_groups_toml(value: &Value) -> String {
let mut groups = Vec::new();
for group in value.as_array().into_iter().flatten() {
let matcher = group
.get("matcher")
.and_then(Value::as_str)
.map(|matcher| format!("matcher={},", toml_string(matcher)))
.unwrap_or_default();
let command = group["hooks"][0]["command"].as_str().unwrap_or_default();
groups.push(format!(
"{{{matcher}hooks=[{{type=\"command\",command={},timeout=30}}]}}",
toml_string(command)
));
}
format!("[{}]", groups.join(","))
}
fn toml_string(value: &str) -> String {
let escaped = value.replace('\\', "\\\\").replace('"', "\\\"");
format!("\"{escaped}\"")
}
fn temp_dir(prefix: &str) -> Result<PathBuf, CliError> {
let path = std::env::temp_dir().join(format!("{prefix}-{}", timestamp()?));
std::fs::create_dir_all(&path)?;
Ok(path)
}
fn cursor_hooks_path() -> Result<PathBuf, CliError> {
let cwd = std::env::current_dir()?;
let project = cwd
.ancestors()
.find(|ancestor| ancestor.join(".cursor").is_dir())
.unwrap_or(cwd.as_path());
Ok(project.join(".cursor/hooks.json"))
}
fn timestamp() -> Result<u128, CliError> {
Ok(SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|error| CliError::Launch(error.to_string()))?
.as_nanos())
}
#[cfg(test)]
#[path = "../tests/coverage/launcher_tests.rs"]
mod tests;