diff --git a/src/loop_cmd.rs b/src/loop_cmd.rs new file mode 100644 index 0000000..aa21d52 --- /dev/null +++ b/src/loop_cmd.rs @@ -0,0 +1,1830 @@ +use std::error::Error; +use std::fmt; +use std::io::{self, IsTerminal, Write}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::{anyhow, bail, Context, Result}; +use braintrust_sdk_rust::{Logs3BatchUploader, SpanComponents, SpanObjectType}; +use clap::{Args, ValueEnum}; +use dialoguer::console::style; +use futures_util::StreamExt; +use indicatif::{ProgressBar, ProgressStyle}; +use reqwest::header::{ACCEPT, ACCEPT_ENCODING}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Map, Value}; +use urlencoding::encode; + +use crate::args::BaseArgs; +use crate::http::{ + build_http_client, build_http_client_from_builder, ApiClient, DEFAULT_HTTP_TIMEOUT, +}; +use crate::project_context::{resolve_project_command_context_with_auth_mode, ProjectContext}; +use crate::ui::{ + animations_enabled, apply_column_padding, header, is_interactive, is_quiet, print_with_pager, + styled_table, truncate, LinePrompt, +}; +use crate::utils::new_uuid_id; + +const DEFAULT_AGENT_SLUG: &str = "loop-chat"; +const OPTIMIZATION_PROJECT_ID: &str = "d341311b-2103-4065-a607-34f2263dd548"; +const LOOP_LOG_ID: &str = "g"; +const LOOP_ROOT_SPAN_NAME: &str = "executor_session"; +const IS_MERGE_FIELD: &str = "_is_merge"; +const BRAINTRUST_PROD_API_URL: &str = "https://api.braintrust.dev"; + +#[derive(Debug, Clone, Args)] +#[command(after_help = "\ +Examples: + bt loop + bt loop --list + bt loop \"Find the most expensive traces from the last day\" + bt loop --conversation daily-debug \"What changed since yesterday?\" + bt loop --harness codex --model gpt-5.4 \"Investigate this project\" + +Interactive commands: + /settings + /harness codex + /model gpt-5.4 + /model clear +")] +pub struct LoopArgs { + /// Message to send. Omit to start an interactive session. + #[arg(value_name = "MESSAGE")] + message: Vec, + + /// Loop Runtime base URL. Defaults to the Braintrust API URL plus /loop/runtime. + #[arg( + long = "runtime-url", + env = "BT_LOOP_RUNTIME_URL", + hide_env_values = true + )] + runtime_url: Option, + + /// Loop agent slug to use or create + #[arg(long, env = "BT_LOOP_AGENT", default_value = DEFAULT_AGENT_SLUG)] + agent: String, + + /// List recent Loop conversations instead of starting a chat + #[arg(long = "list", default_value_t = false)] + list: bool, + + /// Number of conversations to list + #[arg(long, env = "BT_LOOP_LIMIT", default_value_t = 20)] + limit: usize, + + /// Conversation slug or id to resume. Creates a slug if it does not exist. + #[arg(long, short = 'c', env = "BT_LOOP_CONVERSATION")] + conversation: Option, + + /// Name for a newly created conversation + #[arg(long = "conversation-name", env = "BT_LOOP_CONVERSATION_NAME")] + conversation_name: Option, + + /// Initial backend harness to use + #[arg(long, env = "BT_LOOP_HARNESS", value_enum, default_value_t = HarnessArg::Default)] + harness: HarnessArg, + + /// Initial model override + #[arg(long, env = "BT_LOOP_MODEL")] + model: Option, + + /// Braintrust trace parent passed through to Loop Runtime for provider tracing + #[arg( + long = "trace-parent", + env = "BT_LOOP_TRACE_PARENT", + hide_env_values = true + )] + trace_parent: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +enum HarnessArg { + Default, + Codex, + #[value(name = "claude-code")] + ClaudeCode, +} + +impl std::fmt::Display for HarnessArg { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_wire()) + } +} + +impl HarnessArg { + fn as_wire(self) -> &'static str { + match self { + Self::Default => "default", + Self::Codex => "codex", + Self::ClaudeCode => "claude-code", + } + } + + fn valid_values() -> String { + Self::value_variants() + .iter() + .filter_map(|value| value.to_possible_value()) + .map(|value| value.get_name().to_string()) + .collect::>() + .join(", ") + } +} + +#[derive(Debug, Clone)] +struct TurnSettings { + harness: HarnessArg, + model: Option, +} + +impl TurnSettings { + fn from_args(args: &LoopArgs) -> Self { + Self { + harness: args.harness, + model: args.model.clone(), + } + } + + fn label(&self) -> String { + match self.model.as_deref() { + Some(model) => format!("{} | {model}", self.harness), + None => self.harness.to_string(), + } + } + + fn print(&self) { + eprintln!("{} {}", style("settings").dim(), style(self.label()).bold()); + } +} + +pub async fn run(base: BaseArgs, args: LoopArgs) -> Result<()> { + let message = args.message.join(" ").trim().to_string(); + + if args.list && !message.is_empty() { + bail!("MESSAGE cannot be used with --list"); + } + + if base.json { + if message.is_empty() && !args.list { + bail!("MESSAGE is required with --json. Example: bt loop --json \"Summarize this project\""); + } + } else if message.is_empty() && !args.list && !is_interactive() { + bail!( + "MESSAGE is required in non-interactive mode. Example: bt loop \"Summarize this project\"" + ); + } + + let ctx = resolve_project_command_context_with_auth_mode(&base, false).await?; + let runtime_url = resolve_loop_runtime_url(ctx.client.base_url(), args.runtime_url.as_deref())?; + let trace_parent_config = + resolve_loop_trace_parent_config(ctx.client.base_url(), args.trace_parent.as_deref())?; + let client = LoopRuntimeClient::new(runtime_url.as_str(), ctx.client.api_key())?; + let settings = TurnSettings::from_args(&args); + + if args.list { + let conversations = client.list_conversations(&ctx.project.id, &args).await?; + if base.json { + println!("{}", serde_json::to_string(&conversations)?); + } else { + print_conversation_list(&ctx, &conversations)?; + } + return Ok(()); + } + + let mut trace_parent = TraceParentState::new(trace_parent_config); + + if base.json { + let conversation = client.create_conversation(&ctx, &args, &settings).await?; + let report = send_and_collect( + &client, + &ctx, + &conversation, + &settings, + &mut trace_parent, + &message, + true, + ) + .await?; + println!("{}", serde_json::to_string(&report)?); + return Ok(()); + } + + if !message.is_empty() { + let conversation = client.create_conversation(&ctx, &args, &settings).await?; + print_chat_header(&ctx, &conversation, &settings); + send_and_print( + &client, + &ctx, + &conversation, + &settings, + &mut trace_parent, + &message, + ) + .await?; + return Ok(()); + } + + print_project_header(&ctx); + run_interactive_chat(&client, &ctx, &args, &mut trace_parent).await +} + +fn resolve_loop_runtime_url(api_url: &str, explicit_runtime_url: Option<&str>) -> Result { + if let Some(runtime_url) = explicit_runtime_url { + let runtime_url = runtime_url.trim(); + if runtime_url.is_empty() { + bail!("--runtime-url must not be empty"); + } + return Ok(runtime_url.trim_end_matches('/').to_string()); + } + + Ok(format!("{}/loop/runtime", api_url.trim_end_matches('/'))) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum TraceParentConfig { + None, + Explicit(String), + UiLogging { project_id: String }, +} + +#[derive(Debug)] +struct TraceParentState { + config: TraceParentConfig, + logging_session: Option, + logging_failed: bool, +} + +impl TraceParentState { + fn new(config: TraceParentConfig) -> Self { + Self { + config, + logging_session: None, + logging_failed: false, + } + } + + async fn parent_for_turn(&mut self, ctx: &ProjectContext, input: &[Value]) -> Option { + match &self.config { + TraceParentConfig::Explicit(parent) => Some(parent.clone()), + TraceParentConfig::None => None, + TraceParentConfig::UiLogging { project_id } => { + if self.logging_session.is_none() && !self.logging_failed { + match LoopLoggingSession::start(&ctx.client, ctx, project_id, input).await { + Ok(session) => { + self.logging_session = Some(session); + } + Err(err) => { + self.logging_failed = true; + warn_loop_logging_failure(&err, "continuing without a trace parent"); + } + } + } + self.logging_session + .as_ref() + .map(|session| session.parent.clone()) + } + } + } + + async fn update_turn_end(&self, client: &ApiClient) { + if let Some(logging_session) = &self.logging_session { + if let Err(err) = logging_session.update_end_time(client).await { + warn_loop_logging_failure( + &err, + "the Loop turn completed, but the root trace span end time may be stale", + ); + } + } + } +} + +fn resolve_loop_trace_parent_config( + api_url: &str, + explicit_trace_parent: Option<&str>, +) -> Result { + if let Some(trace_parent) = explicit_trace_parent { + let trace_parent = trace_parent.trim(); + if trace_parent.is_empty() { + bail!("--trace-parent must not be empty"); + } + return Ok(TraceParentConfig::Explicit(trace_parent.to_string())); + } + + if should_auto_trace_loop(api_url) { + Ok(TraceParentConfig::UiLogging { + project_id: OPTIMIZATION_PROJECT_ID.to_string(), + }) + } else { + Ok(TraceParentConfig::None) + } +} + +fn should_auto_trace_loop(api_url: &str) -> bool { + let normalized = api_url.trim().trim_end_matches('/'); + normalized.eq_ignore_ascii_case(BRAINTRUST_PROD_API_URL) +} + +async fn run_interactive_chat( + client: &LoopRuntimeClient, + ctx: &ProjectContext, + args: &LoopArgs, + trace_parent: &mut TraceParentState, +) -> Result<()> { + let mut settings = TurnSettings::from_args(args); + let mut initial_history = Vec::new(); + let mut conversation = if args.conversation.is_some() { + let created = client.create_conversation(ctx, args, &settings).await?; + print_conversation_header(&created, &settings); + let events = client + .get_conversation_events(&created.agent.id, &created.conversation.id) + .await?; + print_history(&events.events)?; + initial_history = user_message_history(&events.events); + Some(created) + } else { + None + }; + let mut editor = LinePrompt::new(initial_history); + let prompt = style("You: ").bold().to_string(); + loop { + let Some(input) = editor.read_line(&prompt, "You: ".len())? else { + return Ok(()); + }; + let message = input.trim(); + if message.is_empty() { + continue; + } + if let Some(command) = handle_interactive_command(message, &mut settings)? { + if command == InteractiveCommand::Exit { + return Ok(()); + } + continue; + } + if conversation.is_none() { + let created = client.create_conversation(ctx, args, &settings).await?; + print_conversation_header(&created, &settings); + conversation = Some(created); + } + let conversation = conversation + .as_ref() + .expect("conversation is created before sending a Loop message"); + send_and_print(client, ctx, conversation, &settings, trace_parent, message).await?; + editor.add_history(message); + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum InteractiveCommand { + Continue, + Exit, +} + +fn handle_interactive_command( + message: &str, + settings: &mut TurnSettings, +) -> Result> { + if matches!(message, "/exit" | "/quit" | "exit" | "quit") { + return Ok(Some(InteractiveCommand::Exit)); + } + + let Some(command) = message.strip_prefix('/') else { + return Ok(None); + }; + let mut parts = command.split_whitespace(); + let Some(name) = parts.next() else { + return Ok(None); + }; + match name { + "settings" => { + if parts.next().is_some() { + bail!("usage: /settings"); + } + settings.print(); + } + "harness" => { + let Some(value) = parts.next() else { + eprintln!( + "{} {}", + style("harness").dim(), + style(settings.harness.to_string()).bold() + ); + return Ok(Some(InteractiveCommand::Continue)); + }; + if parts.next().is_some() { + bail!("usage: /harness <{}>", HarnessArg::valid_values()); + } + settings.harness = HarnessArg::from_str(value, true).map_err(|_| { + anyhow::anyhow!( + "unknown harness '{value}'. Use one of: {}", + HarnessArg::valid_values() + ) + })?; + settings.print(); + } + "model" => { + let Some(value) = parts.next() else { + eprintln!( + "{} {}", + style("model").dim(), + style(settings.model.as_deref().unwrap_or("default")).bold() + ); + return Ok(Some(InteractiveCommand::Continue)); + }; + if parts.next().is_some() { + bail!("usage: /model "); + } + if matches!(value, "clear" | "default" | "none" | "reset") { + settings.model = None; + } else { + settings.model = Some(value.to_string()); + } + settings.print(); + } + "help" => { + eprintln!( + "{}\n /settings\n /harness <{}>\n /model \n /exit", + style("commands").dim(), + HarnessArg::valid_values() + ); + } + _ => { + bail!("unknown command '/{name}'. Use /help."); + } + } + Ok(Some(InteractiveCommand::Continue)) +} + +fn print_history(events: &[RuntimeEvent]) -> Result<()> { + for event in events { + match event.event_type() { + Some("messages") => print_history_messages(event)?, + Some("error") => { + let message = event + .data + .get("message") + .and_then(Value::as_str) + .unwrap_or("Loop turn failed"); + eprintln!("{} {}", style("error").red().bold(), message); + } + _ => {} + } + } + Ok(()) +} + +fn print_history_messages(event: &RuntimeEvent) -> Result<()> { + let Some(messages) = event.data.get("messages").and_then(Value::as_array) else { + return Ok(()); + }; + for message in messages { + let text = message_content_text(message.get("content")); + if text.trim().is_empty() { + continue; + } + match message.get("role").and_then(Value::as_str) { + Some("user") => println!("{} {text}", style("You:").bold()), + Some("assistant") => println!("{} {text}", style("Loop:").bold()), + _ => {} + } + } + io::stdout().flush()?; + Ok(()) +} + +fn user_message_history(events: &[RuntimeEvent]) -> Vec { + events + .iter() + .filter_map(|event| event.data.get("messages").and_then(Value::as_array)) + .flat_map(|messages| messages.iter()) + .filter(|message| message.get("role").and_then(Value::as_str) == Some("user")) + .map(|message| message_content_text(message.get("content"))) + .filter(|text| !text.trim().is_empty()) + .collect() +} + +async fn send_and_print( + client: &LoopRuntimeClient, + ctx: &ProjectContext, + conversation: &CreateConversationResponse, + settings: &TurnSettings, + trace_parent: &mut TraceParentState, + message: &str, +) -> Result<()> { + let mut renderer = TranscriptRenderer::default(); + let report = send_and_collect_with_callback( + client, + ctx, + conversation, + settings, + trace_parent, + message, + false, + |event| renderer.render_event(event), + ) + .await?; + renderer.finish_assistant_line()?; + if report.ended_with_error { + bail!("Loop turn failed"); + } + Ok(()) +} + +async fn send_and_collect( + client: &LoopRuntimeClient, + ctx: &ProjectContext, + conversation: &CreateConversationResponse, + settings: &TurnSettings, + trace_parent: &mut TraceParentState, + message: &str, + include_events: bool, +) -> Result { + send_and_collect_with_callback( + client, + ctx, + conversation, + settings, + trace_parent, + message, + include_events, + |_| Ok(()), + ) + .await +} + +async fn send_and_collect_with_callback( + client: &LoopRuntimeClient, + ctx: &ProjectContext, + conversation: &CreateConversationResponse, + settings: &TurnSettings, + trace_parent: &mut TraceParentState, + message: &str, + include_events: bool, + mut on_event: F, +) -> Result +where + F: FnMut(&RuntimeEvent) -> Result<()>, +{ + let input = vec![json!({ + "role": "user", + "content": message, + })]; + let parent = trace_parent.parent_for_turn(ctx, &input).await; + + let submission = client + .submit_turn( + &conversation.agent.id, + &conversation.conversation.id, + parent.as_deref(), + SubmitTurnBody { + input, + harness: settings.harness.as_wire(), + model: settings.model.as_deref(), + }, + ) + .await; + let submission = match submission { + Ok(submission) => submission, + Err(error) => { + trace_parent.update_turn_end(&ctx.client).await; + return Err(error); + } + }; + + let mut status = TurnStatus::new(!include_events, "Waiting for Loop..."); + let mut collected = Vec::new(); + let mut ended_with_error = false; + let watch_result = client + .watch_events( + &submission.agent.id, + &submission.conversation.id, + conversation.conversation.latest_event_id.as_deref(), + |event| { + if event.turn_id.as_deref() != Some(submission.turn.id.as_str()) { + return Ok(true); + } + + if let Some(message) = runtime_status_message(&event) { + status.set_message(message); + } + if event_starts_visible_output(&event) || event.is_turn_ended() { + status.clear(); + } + if event.is_error() { + ended_with_error = true; + } + if event.is_turn_ended() { + if include_events { + collected.push(event); + } + return Ok(false); + } + if runtime_status_message(&event).is_none() { + on_event(&event)?; + } + if include_events { + collected.push(event); + } + Ok(true) + }, + ) + .await; + status.clear(); + trace_parent.update_turn_end(&ctx.client).await; + watch_result?; + + Ok(LoopChatReport { + submission, + events: collected, + ended_with_error, + }) +} + +#[derive(Debug, Clone)] +struct LoopLoggingSession { + id: String, + project_id: String, + root_span_id: String, + span_id: String, + parent: String, +} + +impl LoopLoggingSession { + async fn start( + client: &ApiClient, + ctx: &ProjectContext, + project_id: &str, + input: &[Value], + ) -> Result { + let id = new_uuid_id(); + let span_id = new_uuid_id(); + let root_span_id = span_id.clone(); + let parent = loop_logging_parent(project_id, &id, &root_span_id, &span_id); + let mut event = json!({ + "id": id, + "root_span_id": root_span_id, + "span_id": span_id, + "span_parents": [], + "span_attributes": { + "name": LOOP_ROOT_SPAN_NAME, + "type": "task", + }, + "input": input, + "metadata": { + "project_id": ctx.project.id, + "org_id": ctx.project.org_id, + "page": "bt_loop", + }, + "metrics": { + "start": unix_timestamp_seconds(), + }, + "project_id": project_id, + "log_id": LOOP_LOG_ID, + }); + event[IS_MERGE_FIELD] = Value::Bool(false); + insert_loop_logging_event(client, event).await.map_err(|err| { + anyhow!("failed to initialize Loop logging span for trace project '{project_id}': {err}") + })?; + + Ok(Self { + id, + project_id: project_id.to_string(), + root_span_id, + span_id, + parent, + }) + } + + async fn update_end_time(&self, client: &ApiClient) -> Result<()> { + let mut event = json!({ + "id": self.id, + "root_span_id": self.root_span_id, + "span_id": self.span_id, + "span_parents": [], + "metrics": { + "end": unix_timestamp_seconds(), + }, + "project_id": self.project_id, + "log_id": LOOP_LOG_ID, + }); + event[IS_MERGE_FIELD] = Value::Bool(true); + insert_loop_logging_event(client, event) + .await + .map_err(|err| { + anyhow!( + "failed to update Loop logging span end time for trace project '{}': {err}", + self.project_id + ) + }) + } +} + +fn warn_loop_logging_failure(err: &anyhow::Error, consequence: &str) { + if is_quiet() { + return; + } + eprintln!("{} {err}; {consequence}.", style("warning:").yellow()); +} + +fn loop_logging_parent(project_id: &str, id: &str, root_span_id: &str, span_id: &str) -> String { + let mut parent = SpanComponents::new(SpanObjectType::ProjectLogs); + parent.object_id = Some(project_id.to_string()); + parent.row_id = Some(id.to_string()); + parent.root_span_id = Some(root_span_id.to_string()); + parent.span_id = Some(span_id.to_string()); + parent.to_str() +} + +async fn insert_loop_logging_event(client: &ApiClient, event: Value) -> Result<()> { + let Value::Object(row) = event else { + bail!("Loop logging event must be a JSON object"); + }; + insert_loop_logging_rows(client, &[row]).await +} + +async fn insert_loop_logging_rows(client: &ApiClient, rows: &[Map]) -> Result<()> { + let mut uploader = Logs3BatchUploader::new( + client.base_url(), + client.api_key().to_string(), + Some(client.org_name().to_string()), + )?; + uploader.upload_rows(rows, 100).await?; + Ok(()) +} + +fn unix_timestamp_seconds() -> f64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs_f64()) + .unwrap_or_default() +} + +fn print_chat_header( + ctx: &ProjectContext, + conversation: &CreateConversationResponse, + settings: &TurnSettings, +) { + print_project_header(ctx); + print_conversation_header(conversation, settings); +} + +fn print_project_header(ctx: &ProjectContext) { + if is_quiet() { + return; + } + eprintln!( + "{} {} {} {}", + style("Loop").bold(), + style("->").dim(), + style(ctx.project.name.as_str()).bold(), + style(format!("({})", ctx.project.id)).dim() + ); +} + +fn print_conversation_header(conversation: &CreateConversationResponse, settings: &TurnSettings) { + if is_quiet() { + return; + } + eprintln!( + "{} {} {}", + style("Conversation").dim(), + style(conversation.conversation.slug.as_str()).bold(), + style(format!("[{}]", settings.label())).dim() + ); +} + +fn print_conversation_list( + ctx: &ProjectContext, + response: &ListConversationsResponse, +) -> Result<()> { + let mut output = String::new(); + if response.conversations.is_empty() { + output.push_str("No Loop conversations found.\n"); + print_with_pager(&output)?; + return Ok(()); + } + + output.push_str( + format!( + "{} {} {} {} {}\n\n", + style(response.conversations.len()).bold(), + style("Loop conversations in").dim(), + style(ctx.project.name.as_str()).bold(), + style("for agent").dim(), + style(response.agent.slug.as_str()).bold() + ) + .as_str(), + ); + + let mut table = styled_table(); + table.set_header(vec![ + header("Name"), + header("Slug"), + header("ID"), + header("Latest event"), + ]); + apply_column_padding(&mut table, (0, 4)); + for conversation in &response.conversations { + let latest_event = conversation + .latest_event_id + .as_deref() + .map(|id| truncate(id, 12)) + .unwrap_or_else(|| "-".to_string()); + table.add_row(vec![ + truncate(&conversation.name, 28), + truncate(&conversation.slug, 28), + truncate(&conversation.id, 12), + latest_event, + ]); + } + output.push_str(&table.to_string()); + print_with_pager(&output)?; + Ok(()) +} + +#[derive(Clone)] +struct LoopRuntimeClient { + http: reqwest::Client, + watch_http: reqwest::Client, + base_url: String, + api_key: String, +} + +impl LoopRuntimeClient { + fn new(base_url: &str, api_key: &str) -> Result { + if base_url.trim().is_empty() { + bail!("--runtime-url must not be empty"); + } + Ok(Self { + http: build_http_client(DEFAULT_HTTP_TIMEOUT)?, + watch_http: build_http_client_from_builder( + reqwest::Client::builder().connect_timeout(DEFAULT_HTTP_TIMEOUT), + )?, + base_url: base_url.trim_end_matches('/').to_string(), + api_key: api_key.to_string(), + }) + } + + async fn create_conversation( + &self, + ctx: &ProjectContext, + args: &LoopArgs, + settings: &TurnSettings, + ) -> Result { + let agent = self.agent_by_slug(&ctx.project.id, &args.agent).await?; + let conversation_id = args.conversation.as_deref().filter(|value| is_uuid(value)); + let conversation_slug = args.conversation.as_deref().filter(|value| !is_uuid(value)); + self.post( + &format!("/agent/{}/conversation", encode(&agent.id)), + &json!({ + "conversation_id": conversation_id, + "conversation_slug": conversation_slug, + "conversation_name": args.conversation_name.as_deref(), + "harness": settings.harness.as_wire(), + }), + ) + .await + .map_err(|err| { + LoopCommandError::new( + format!( + "failed to create or resolve Loop conversation for project '{}'", + ctx.project.name + ), + err, + ) + .into() + }) + } + + async fn list_conversations( + &self, + project_id: &str, + args: &LoopArgs, + ) -> Result { + let agent = self.agent_by_slug(project_id, &args.agent).await?; + let response: ListConversationsResponse = self + .get(&format!( + "/agent/{}/conversation?limit={}", + encode(&agent.id), + args.limit, + )) + .await + .map_err(|err| LoopCommandError::new("failed to list Loop conversations", err))?; + Ok(response) + } + + async fn get_conversation_events( + &self, + agent_id: &str, + conversation_id: &str, + ) -> Result { + let response = self + .get(&format!( + "/agent/{}/conversation/{}/event", + encode(agent_id), + encode(conversation_id), + )) + .await + .map_err(|err| LoopCommandError::new("failed to load Loop conversation", err))?; + Ok(response) + } + + async fn agent_by_slug(&self, project_id: &str, slug: &str) -> Result { + let response: ListAgentsResponse = self + .get(&format!( + "/agent?project_id={}&slug={}", + encode(project_id), + encode(slug), + )) + .await + .map_err(|err| LoopCommandError::new("failed to list Loop agents", err))?; + response + .agents + .into_iter() + .find(|agent| agent.slug == slug) + .ok_or_else(|| { + LoopCommandError::new( + "failed to list Loop conversations", + LoopRuntimeRequestError::new(format!("Loop agent not found: {slug}")).into(), + ) + .into() + }) + } + + async fn submit_turn( + &self, + agent_id: &str, + conversation_id: &str, + parent: Option<&str>, + body: SubmitTurnBody<'_>, + ) -> Result { + self.post( + &format!( + "/agent/{}/conversation/{}/turn", + encode(agent_id), + encode(conversation_id) + ), + &SubmitTurnBodyWithParent { parent, body }, + ) + .await + .map_err(|err| LoopCommandError::new("failed to submit Loop turn", err).into()) + } + + async fn watch_events( + &self, + agent_id: &str, + conversation_id: &str, + after: Option<&str>, + mut on_event: F, + ) -> Result<()> + where + F: FnMut(RuntimeEvent) -> Result, + { + let mut url = reqwest::Url::parse(&self.url(&format!( + "/agent/{}/conversation/{}/event/watch", + encode(agent_id), + encode(conversation_id) + )))?; + if let Some(after) = after { + url.query_pairs_mut().append_pair("after", after); + } + let url_string = url.to_string(); + let response = self + .watch_http + .get(url) + .bearer_auth(&self.api_key) + .header(ACCEPT, "text/event-stream") + .header(ACCEPT_ENCODING, "identity") + .send() + .await + .map_err(|err| { + LoopCommandError::new( + "failed to watch Loop events", + LoopRuntimeRequestError::with_source( + format!("Loop Runtime request failed: GET {url_string}"), + err, + ) + .into(), + ) + })?; + let status = response.status(); + if !status.is_success() { + let body = response.bytes().await.map_err(|err| { + LoopCommandError::new( + "failed to watch Loop events", + LoopRuntimeRequestError::with_source( + format!("failed to read Loop Runtime response body from GET {url_string}"), + err, + ) + .into(), + ) + })?; + let body_text = String::from_utf8_lossy(&body); + return Err(LoopCommandError::new( + "failed to watch Loop events", + LoopRuntimeRequestError::new(format!( + "Loop Runtime request failed: GET {url_string} returned {status}: {body_text}" + )) + .into(), + ) + .into()); + } + let mut stream = response.bytes_stream(); + let mut buffer = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|err| { + LoopCommandError::new( + "failed to watch Loop events", + LoopRuntimeRequestError::with_source( + format!("Loop Runtime event stream failed: GET {url_string}"), + err, + ) + .into(), + ) + })?; + buffer.extend_from_slice(&chunk); + while let Some((boundary, separator_len)) = sse_event_boundary(&buffer) { + let raw_event = buffer.drain(..boundary).collect::>(); + buffer.drain(..separator_len); + if let Some(event) = parse_sse_event(&raw_event)? { + match event.name.as_str() { + "exo_event" => { + let runtime_event = serde_json::from_str::(&event.data) + .with_context(|| { + format!("failed to parse Loop Runtime event from {url_string}") + })?; + if !on_event(runtime_event)? { + return Ok(()); + } + } + "error" => { + bail!("Loop Runtime event stream failed: {}", event.data); + } + _ => {} + } + } + } + } + Ok(()) + } + + async fn post(&self, path: &str, body: &B) -> Result + where + T: for<'de> Deserialize<'de>, + B: Serialize, + { + let url = self.url(path); + let response = self + .http + .post(&url) + .bearer_auth(&self.api_key) + .json(body) + .send() + .await + .map_err(|err| { + LoopRuntimeRequestError::with_source( + format!("Loop Runtime request failed: POST {url}"), + err, + ) + })?; + parse_loop_runtime_response(response, "POST", &url).await + } + + async fn get(&self, path: &str) -> Result + where + T: for<'de> Deserialize<'de>, + { + let url = self.url(path); + let response = self + .http + .get(&url) + .bearer_auth(&self.api_key) + .send() + .await + .map_err(|err| { + LoopRuntimeRequestError::with_source( + format!("Loop Runtime request failed: GET {url}"), + err, + ) + })?; + parse_loop_runtime_response(response, "GET", &url).await + } + + fn url(&self, path: &str) -> String { + format!("{}/{}", self.base_url, path.trim_start_matches('/')) + } +} + +#[derive(Debug)] +struct LoopCommandError { + message: String, + source: anyhow::Error, +} + +impl LoopCommandError { + fn new(message: impl Into, source: anyhow::Error) -> Self { + Self { + message: message.into(), + source, + } + } +} + +impl fmt::Display for LoopCommandError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}: {}", self.message, self.source) + } +} + +impl Error for LoopCommandError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + Some(self.source.as_ref()) + } +} + +#[derive(Debug)] +struct LoopRuntimeRequestError { + message: String, + source: Option, +} + +impl LoopRuntimeRequestError { + fn new(message: impl Into) -> Self { + Self { + message: message.into(), + source: None, + } + } + + fn with_source(message: impl Into, source: reqwest::Error) -> Self { + Self { + message: message.into(), + source: Some(source), + } + } +} + +impl fmt::Display for LoopRuntimeRequestError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match &self.source { + Some(source) => write!(f, "{}: {source}", self.message), + None => f.write_str(&self.message), + } + } +} + +impl Error for LoopRuntimeRequestError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + self.source + .as_ref() + .map(|err| err as &(dyn Error + 'static)) + } +} + +async fn parse_loop_runtime_response( + response: reqwest::Response, + method: &str, + url: &str, +) -> Result +where + T: for<'de> Deserialize<'de>, +{ + let status = response.status(); + let body = response.bytes().await.map_err(|err| { + LoopRuntimeRequestError::with_source( + format!("failed to read Loop Runtime response body from {method} {url}"), + err, + ) + })?; + if !status.is_success() { + let body_text = String::from_utf8_lossy(&body); + return Err(LoopRuntimeRequestError::new(format!( + "Loop Runtime request failed: {method} {url} returned {status}: {body_text}" + )) + .into()); + } + serde_json::from_slice(&body) + .with_context(|| format!("failed to parse Loop Runtime response from {method} {url}")) +} + +struct SseEvent { + name: String, + data: String, +} + +fn sse_event_boundary(buffer: &[u8]) -> Option<(usize, usize)> { + if let Some(index) = buffer.windows(4).position(|window| window == b"\r\n\r\n") { + return Some((index, 4)); + } + buffer + .windows(2) + .position(|window| window == b"\n\n") + .map(|index| (index, 2)) +} + +fn parse_sse_event(raw_event: &[u8]) -> Result> { + if raw_event.is_empty() { + return Ok(None); + } + let text = std::str::from_utf8(raw_event).context("failed to parse Loop Runtime SSE frame")?; + let mut name = "message".to_string(); + let mut data = Vec::new(); + for raw_line in text.lines() { + let line = raw_line.trim_end_matches('\r'); + if let Some(value) = line.strip_prefix("event:") { + name = value.trim().to_string(); + } else if let Some(value) = line.strip_prefix("data:") { + data.push(value.trim_start().to_string()); + } + } + Ok(Some(SseEvent { + name, + data: data.join("\n"), + })) +} + +struct TurnStatus { + spinner: Option, +} + +impl TurnStatus { + fn new(enabled: bool, message: &str) -> Self { + if !enabled || !io::stderr().is_terminal() || !animations_enabled() || is_quiet() { + return Self { spinner: None }; + } + let spinner = ProgressBar::new_spinner(); + spinner.set_style( + ProgressStyle::default_spinner() + .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏", " "]) + .template("{spinner:.cyan} {msg}") + .expect("spinner template should be valid"), + ); + spinner.set_message(message.to_string()); + spinner.enable_steady_tick(Duration::from_millis(80)); + Self { + spinner: Some(spinner), + } + } + + fn set_message(&self, message: &str) { + if let Some(spinner) = &self.spinner { + spinner.set_message(message.to_string()); + } + } + + fn clear(&mut self) { + if let Some(spinner) = self.spinner.take() { + spinner.finish_and_clear(); + } + } +} + +impl Drop for TurnStatus { + fn drop(&mut self) { + self.clear(); + } +} + +#[derive(Debug, Serialize)] +struct LoopChatReport { + submission: SubmitTurnResponse, + events: Vec, + ended_with_error: bool, +} + +#[derive(Debug, Deserialize)] +struct ListAgentsResponse { + agents: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +struct ListConversationsResponse { + agent: AgentRecord, + conversations: Vec, + next_cursor: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +struct GetConversationEventsResponse { + events: Vec, + cursor: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +struct AgentRecord { + id: String, + slug: String, + name: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct ConversationRecord { + id: String, + slug: String, + name: String, + #[serde(default)] + latest_event_id: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +struct TurnRecord { + id: String, + session_id: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct CreateConversationResponse { + agent: AgentRecord, + conversation: ConversationRecord, + harness: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct SubmitTurnResponse { + agent: AgentRecord, + conversation: ConversationRecord, + turn: TurnRecord, + harness: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct RuntimeEvent { + id: String, + #[serde(default)] + turn_id: Option, + data: Value, +} + +impl RuntimeEvent { + fn event_type(&self) -> Option<&str> { + self.data.get("type").and_then(Value::as_str) + } + + fn is_turn_ended(&self) -> bool { + self.event_type() == Some("turn_ended") + } + + fn is_error(&self) -> bool { + self.event_type() == Some("error") + } +} + +fn runtime_status_message(event: &RuntimeEvent) -> Option<&str> { + if event.event_type() != Some("custom") { + return None; + } + if event.data.get("event_type").and_then(Value::as_str) != Some("agent_runtime.status") { + return None; + } + event + .data + .pointer("/payload/message") + .and_then(Value::as_str) +} + +fn event_starts_visible_output(event: &RuntimeEvent) -> bool { + match event.event_type() { + Some("lingua_stream_chunk") => !stream_chunk_text(event).is_empty(), + Some("messages") => event + .data + .get("messages") + .and_then(Value::as_array) + .is_some_and(|messages| { + messages.iter().any(|message| { + message.get("role").and_then(Value::as_str) == Some("assistant") + && !message_content_text(message.get("content")) + .trim() + .is_empty() + }) + }), + Some("tool_requested" | "tool_result" | "error") => true, + _ => false, + } +} + +#[derive(Debug, Serialize)] +struct SubmitTurnBody<'a> { + input: Vec, + harness: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + model: Option<&'a str>, +} + +#[derive(Debug, Serialize)] +struct SubmitTurnBodyWithParent<'a> { + #[serde(skip_serializing_if = "Option::is_none")] + parent: Option<&'a str>, + #[serde(flatten)] + body: SubmitTurnBody<'a>, +} + +#[derive(Default)] +struct TranscriptRenderer { + assistant_line_open: bool, + streamed_assistant_text: bool, + rendered_assistant_message: bool, +} + +impl TranscriptRenderer { + fn render_event(&mut self, event: &RuntimeEvent) -> Result<()> { + match event.event_type() { + Some("lingua_stream_chunk") => self.render_stream_chunk(event), + Some("messages") => self.render_messages(event), + Some("tool_requested") => self.render_tool_request(event), + Some("tool_result") => self.render_tool_result(event), + Some("custom") => self.render_custom(event), + Some("error") => self.render_error(event), + _ => Ok(()), + } + } + + fn render_stream_chunk(&mut self, event: &RuntimeEvent) -> Result<()> { + let text = stream_chunk_text(event); + if text.is_empty() { + return Ok(()); + } + self.open_assistant_line()?; + print!("{text}"); + io::stdout().flush()?; + self.streamed_assistant_text = true; + Ok(()) + } + + fn render_messages(&mut self, event: &RuntimeEvent) -> Result<()> { + if self.streamed_assistant_text || self.rendered_assistant_message { + return Ok(()); + } + let Some(messages) = event.data.get("messages").and_then(Value::as_array) else { + return Ok(()); + }; + for message in messages { + if message.get("role").and_then(Value::as_str) != Some("assistant") { + continue; + } + let text = message_content_text(message.get("content")); + if text.trim().is_empty() { + continue; + } + self.open_assistant_line()?; + print!("{text}"); + self.rendered_assistant_message = true; + } + io::stdout().flush()?; + Ok(()) + } + + fn render_tool_request(&mut self, event: &RuntimeEvent) -> Result<()> { + self.finish_assistant_line()?; + let function_name = event + .data + .pointer("/request/function_name") + .and_then(Value::as_str) + .unwrap_or("tool"); + eprintln!("{} {}", style("tool").dim(), style(function_name).cyan()); + Ok(()) + } + + fn render_tool_result(&mut self, event: &RuntimeEvent) -> Result<()> { + self.finish_assistant_line()?; + let tool_call_id = event + .data + .get("tool_call_id") + .and_then(Value::as_str) + .unwrap_or("tool"); + eprintln!( + "{} {}", + style("tool result").dim(), + style(tool_call_id).dim() + ); + Ok(()) + } + + fn render_custom(&mut self, event: &RuntimeEvent) -> Result<()> { + if event.data.get("event_type").and_then(Value::as_str) != Some("agent_runtime.status") { + return Ok(()); + } + let Some(message) = event + .data + .pointer("/payload/message") + .and_then(Value::as_str) + else { + return Ok(()); + }; + self.finish_assistant_line()?; + eprintln!("{}", style(message).dim()); + Ok(()) + } + + fn render_error(&mut self, event: &RuntimeEvent) -> Result<()> { + self.finish_assistant_line()?; + let message = event + .data + .get("message") + .and_then(Value::as_str) + .unwrap_or("Loop turn failed"); + eprintln!("{} {}", style("error").red().bold(), message); + Ok(()) + } + + fn open_assistant_line(&mut self) -> Result<()> { + if self.assistant_line_open { + return Ok(()); + } + print!("{}", style("Loop: ").bold()); + io::stdout().flush()?; + self.assistant_line_open = true; + Ok(()) + } + + fn finish_assistant_line(&mut self) -> Result<()> { + if self.assistant_line_open { + println!(); + io::stdout().flush()?; + self.assistant_line_open = false; + } + Ok(()) + } +} + +fn stream_chunk_text(event: &RuntimeEvent) -> String { + let Some(choices) = event + .data + .pointer("/chunk/choices") + .and_then(Value::as_array) + else { + return String::new(); + }; + choices + .iter() + .filter_map(|choice| { + choice + .pointer("/delta/content") + .and_then(Value::as_str) + .map(str::to_string) + }) + .collect::>() + .join("") +} + +fn message_content_text(content: Option<&Value>) -> String { + match content { + Some(Value::String(text)) => text.clone(), + Some(Value::Array(parts)) => parts + .iter() + .filter_map(|part| part.get("text").and_then(Value::as_str)) + .collect::>() + .join(""), + _ => String::new(), + } +} + +fn is_uuid(value: &str) -> bool { + let bytes = value.as_bytes(); + if bytes.len() != 36 { + return false; + } + for (index, byte) in bytes.iter().enumerate() { + if matches!(index, 8 | 13 | 18 | 23) { + if *byte != b'-' { + return false; + } + } else if !byte.is_ascii_hexdigit() { + return false; + } + } + true +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn uuid_detection_accepts_canonical_ids() { + assert!(is_uuid("123e4567-e89b-12d3-a456-426614174000")); + assert!(!is_uuid("daily-debug")); + } + + #[test] + fn default_agent_slug_matches_loop_chat_runtime_agent() { + assert_eq!(DEFAULT_AGENT_SLUG, "loop-chat"); + } + + #[test] + fn loop_runtime_url_defaults_to_api_proxy_path() { + assert_eq!( + resolve_loop_runtime_url("http://localhost:8000/", None).expect("runtime URL"), + "http://localhost:8000/loop/runtime" + ); + } + + #[test] + fn loop_runtime_url_uses_explicit_override() { + assert_eq!( + resolve_loop_runtime_url( + "http://localhost:8000", + Some(" https://loop-runtime.example.test/ "), + ) + .expect("runtime URL"), + "https://loop-runtime.example.test" + ); + } + + #[test] + fn loop_trace_parent_auto_matches_braintrust_prod_api_url() { + assert_eq!( + resolve_loop_trace_parent_config("https://api.braintrust.dev/", None) + .expect("trace parent"), + TraceParentConfig::UiLogging { + project_id: OPTIMIZATION_PROJECT_ID.to_string() + } + ); + } + + #[test] + fn loop_trace_parent_auto_omits_non_prod_api_urls() { + assert_eq!( + resolve_loop_trace_parent_config("https://api-eu.braintrust.dev", None) + .expect("trace parent"), + TraceParentConfig::None + ); + assert_eq!( + resolve_loop_trace_parent_config("https://staging-api.braintrust.dev", None) + .expect("trace parent"), + TraceParentConfig::None + ); + assert_eq!( + resolve_loop_trace_parent_config("https://runtime.example.test", None) + .expect("trace parent"), + TraceParentConfig::None + ); + } + + #[test] + fn loop_trace_parent_explicit_override_wins() { + assert_eq!( + resolve_loop_trace_parent_config( + "https://runtime.example.test", + Some(" project_name:test-project "), + ) + .expect("trace parent"), + TraceParentConfig::Explicit("project_name:test-project".to_string()) + ); + } + + #[test] + fn loop_trace_parent_explicit_override_accepts_span_components() { + let parent = loop_logging_parent( + "custom-project-id", + "123e4567-e89b-12d3-a456-426614174000", + "root-span-id", + "span-id", + ); + let padded_parent = format!(" {parent} "); + assert_eq!( + resolve_loop_trace_parent_config("https://runtime.example.test", Some(&padded_parent)) + .expect("trace parent"), + TraceParentConfig::Explicit(parent) + ); + } + + #[test] + fn loop_logging_parent_round_trips_as_project_logs_span() { + let parent = loop_logging_parent( + "custom-project-id", + "123e4567-e89b-12d3-a456-426614174000", + "root-span-id", + "span-id", + ); + let parsed = SpanComponents::parse(&parent).expect("span parent"); + assert_eq!(parsed.object_type, SpanObjectType::ProjectLogs); + assert_eq!(parsed.object_id.as_deref(), Some("custom-project-id")); + assert_eq!( + parsed.row_id.as_deref(), + Some("123e4567-e89b-12d3-a456-426614174000") + ); + assert_eq!(parsed.root_span_id.as_deref(), Some("root-span-id")); + assert_eq!(parsed.span_id.as_deref(), Some("span-id")); + } + + #[test] + fn interactive_command_switches_harness_and_model() { + let mut settings = TurnSettings { + harness: HarnessArg::Default, + model: None, + }; + + assert_eq!( + handle_interactive_command("/harness codex", &mut settings).expect("harness command"), + Some(InteractiveCommand::Continue) + ); + assert_eq!(settings.harness, HarnessArg::Codex); + + assert_eq!( + handle_interactive_command("/model gpt-5.4", &mut settings).expect("model command"), + Some(InteractiveCommand::Continue) + ); + assert_eq!(settings.model.as_deref(), Some("gpt-5.4")); + assert_eq!(settings.label(), "codex | gpt-5.4"); + } + + #[test] + fn interactive_command_clears_model_and_detects_exit() { + let mut settings = TurnSettings { + harness: HarnessArg::ClaudeCode, + model: Some("claude-opus-4-5".to_string()), + }; + + assert_eq!( + handle_interactive_command("/model clear", &mut settings).expect("model clear"), + Some(InteractiveCommand::Continue) + ); + assert_eq!(settings.model, None); + + assert_eq!( + handle_interactive_command("/exit", &mut settings).expect("exit command"), + Some(InteractiveCommand::Exit) + ); + } + + #[test] + fn stream_chunk_text_reads_openai_style_delta_content() { + let event = RuntimeEvent { + id: "event-id".to_string(), + turn_id: Some("turn-id".to_string()), + data: json!({ + "type": "lingua_stream_chunk", + "chunk": { + "choices": [ + {"index": 0, "delta": {"content": "hello "}}, + {"index": 0, "delta": {"content": "world"}} + ] + } + }), + }; + assert_eq!(stream_chunk_text(&event), "hello world"); + } + + #[test] + fn sse_boundary_reads_lf_and_crlf_frames() { + assert_eq!(sse_event_boundary(b"event: exo_event\n\n"), Some((16, 2))); + assert_eq!( + sse_event_boundary(b"event: exo_event\r\n\r\n"), + Some((16, 4)) + ); + } + + #[test] + fn parse_sse_event_reads_runtime_frame() { + let event = parse_sse_event(b"event: exo_event\r\ndata: {\"id\":\"event-id\"}\r\n") + .expect("valid frame") + .expect("event"); + assert_eq!(event.name, "exo_event"); + assert_eq!(event.data, "{\"id\":\"event-id\"}"); + } + + #[test] + fn message_content_text_reads_string_and_text_parts() { + assert_eq!( + message_content_text(Some(&json!("plain response"))), + "plain response" + ); + assert_eq!( + message_content_text(Some(&json!([ + {"type": "text", "text": "part one"}, + {"type": "text", "text": " part two"} + ]))), + "part one part two" + ); + } + + #[test] + fn user_message_history_reads_only_conversation_user_messages() { + let events = vec![ + RuntimeEvent { + id: "event-1".to_string(), + turn_id: Some("turn-1".to_string()), + data: json!({ + "type": "messages", + "messages": [ + {"role": "user", "content": "first question"}, + {"role": "assistant", "content": "first answer"} + ] + }), + }, + RuntimeEvent { + id: "event-2".to_string(), + turn_id: Some("turn-2".to_string()), + data: json!({ + "type": "messages", + "messages": [ + {"role": "user", "content": [{"type": "text", "text": "second question"}]} + ] + }), + }, + ]; + + assert_eq!( + user_message_history(&events), + vec!["first question".to_string(), "second question".to_string()] + ); + } +} diff --git a/src/main.rs b/src/main.rs index a4780fc..3c25dbd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,6 +15,7 @@ mod functions; mod http; mod init; mod js_runner; +mod loop_cmd; mod project_context; mod projects; mod prompts; @@ -70,6 +71,7 @@ Projects & resources tools Manage tools scorers Manage scorers experiments Manage experiments + loop Chat with Loop for the active project Data & evaluation datasets Manage datasets @@ -155,6 +157,8 @@ enum Commands { Functions(CLIArgs), /// Manage experiments Experiments(CLIArgs), + /// Chat with Loop for the active project + Loop(CLIArgs), /// Synchronize project logs between Braintrust and local NDJSON files Sync(CLIArgs), /// Local utility commands @@ -187,6 +191,7 @@ impl Commands { Commands::Scorers(cmd) => &cmd.base, Commands::Functions(cmd) => &cmd.base, Commands::Experiments(cmd) => &cmd.base, + Commands::Loop(cmd) => &cmd.base, Commands::Sync(cmd) => &cmd.base, Commands::Util(cmd) => &cmd.base, Commands::Switch(cmd) => &cmd.base, @@ -213,6 +218,7 @@ impl Commands { Commands::Scorers(cmd) => &mut cmd.base, Commands::Functions(cmd) => &mut cmd.base, Commands::Experiments(cmd) => &mut cmd.base, + Commands::Loop(cmd) => &mut cmd.base, Commands::Sync(cmd) => &mut cmd.base, Commands::Util(cmd) => &mut cmd.base, Commands::Switch(cmd) => &mut cmd.base, @@ -319,6 +325,7 @@ fn try_main() -> Result<()> { Commands::Scorers(cmd) => scorers::run(cmd.base, cmd.args).await?, Commands::Functions(cmd) => functions::run(cmd.base, cmd.args).await?, Commands::Experiments(cmd) => experiments::run(cmd.base, cmd.args).await?, + Commands::Loop(cmd) => loop_cmd::run(cmd.base, cmd.args).await?, Commands::Sync(cmd) => sync::run(cmd.base, cmd.args).await?, Commands::Util(cmd) => util_cmd::run(cmd.base, cmd.args).await?, Commands::SelfCommand(cmd) => self_update::run(cmd.base, cmd.args).await?, diff --git a/src/sql.rs b/src/sql.rs index 208535a..132c969 100644 --- a/src/sql.rs +++ b/src/sql.rs @@ -5,18 +5,7 @@ use std::time::{Duration, Instant}; use anyhow::{bail, Context, Result}; use clap::{builder::BoolishValueParser, Args}; -use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyModifiers}; -use crossterm::terminal::{ - disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen, -}; -use crossterm::ExecutableCommand; -use ratatui::backend::CrosstermBackend; -use ratatui::layout::{Constraint, Direction, Layout, Rect}; -use ratatui::prelude::Frame; -use ratatui::style::Style; -use ratatui::text::Line; -use ratatui::widgets::{Block, Borders, Paragraph, Wrap}; -use ratatui::Terminal; +use dialoguer::console::style; use serde::{Deserialize, Serialize}; use serde_json::{json, Map, Value}; use unicode_width::UnicodeWidthStr; @@ -25,7 +14,7 @@ use urlencoding::encode; use crate::args::BaseArgs; use crate::auth::login; use crate::http::{ApiClient, HttpError}; -use crate::ui::with_spinner; +use crate::ui::{with_spinner, LinePrompt}; use crate::utils::parse_duration_to_seconds; const QUERY_SOURCE: &str = "bt_sql_9f4b1e6d7c2a4a7b8d4f9a6c2b1e7f3d"; @@ -286,145 +275,31 @@ async fn run_interactive(base: BaseArgs, client: ApiClient, lint_mode: String) - tokio::task::block_in_place(|| run_interactive_blocking(base.json, client, handle, lint_mode)) } -struct TerminalGuard; - -impl Drop for TerminalGuard { - fn drop(&mut self) { - disable_raw_mode().ok(); - io::stdout().execute(LeaveAlternateScreen).ok(); - } -} - fn run_interactive_blocking( json_output: bool, client: ApiClient, handle: tokio::runtime::Handle, lint_mode: String, ) -> Result<()> { - enable_raw_mode()?; - let _guard = TerminalGuard; - let mut stdout = io::stdout(); - stdout.execute(EnterAlternateScreen)?; - let backend = CrosstermBackend::new(stdout); - let mut terminal = Terminal::new(backend)?; - - let res = run_app(&mut terminal, json_output, client, handle, lint_mode); - terminal.show_cursor().ok(); - res -} - -fn run_app( - terminal: &mut Terminal>, - json_output: bool, - client: ApiClient, - handle: tokio::runtime::Handle, - lint_mode: String, -) -> Result<()> { - let mut app = App::new(json_output, lint_mode); + let mut editor = LinePrompt::new(Vec::new()); + let prompt = style("SQL> ").bold().to_string(); loop { - terminal.draw(|f| ui(f, &app))?; - - if event::poll(Duration::from_millis(200))? { - match event::read()? { - Event::Key(key) => { - if handle_key_event(&mut app, key, &client, &handle)? { - break; - } - } - Event::Resize(_, _) => {} - _ => {} - } + let Some(input) = editor.read_line(&prompt, "SQL> ".len())? else { + return Ok(()); + }; + let query = input.trim(); + if query.is_empty() { + continue; } - } - Ok(()) -} - -fn handle_key_event( - app: &mut App, - key: KeyEvent, - client: &ApiClient, - handle: &tokio::runtime::Handle, -) -> Result { - match key.code { - KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => { - app.clear_input(); - app.status = "Cleared input".to_string(); - } - KeyCode::Char('d') if key.modifiers.contains(KeyModifiers::CONTROL) => return Ok(true), - KeyCode::Esc => return Ok(true), - KeyCode::Char('l') if key.modifiers.contains(KeyModifiers::CONTROL) => { - app.output.clear(); + match handle.block_on(execute_query(&client, query, lint_mode.as_str())) { + Ok(response) => print_response(&response, json_output)?, + Err(err) => eprintln!("{} {err}", style("error").red().bold()), } - KeyCode::Enter => { - let query = app.input.trim().to_string(); - if query.is_empty() { - return Ok(false); - } - - app.status = "Running query...".to_string(); - let result = handle.block_on(execute_query(client, &query, app.lint_mode.as_str())); - match result { - Ok(response) => { - app.output = format_response(&response, app.json_output)?; - app.status = "OK".to_string(); - } - Err(err) => { - app.output = format!("Error: {err}"); - app.status = "Error".to_string(); - } - } - app.push_history(&query); - app.clear_input(); - } - KeyCode::Backspace => app.backspace(), - KeyCode::Delete => app.delete(), - KeyCode::Left => app.move_left(), - KeyCode::Right => app.move_right(), - KeyCode::Home => app.move_home(), - KeyCode::End => app.move_end(), - KeyCode::Up => app.history_prev(), - KeyCode::Down => app.history_next(), - KeyCode::Char(ch) - if !key.modifiers.contains(KeyModifiers::CONTROL) - && !key.modifiers.contains(KeyModifiers::ALT) => - { - app.insert_char(ch); - } - _ => {} + editor.add_history(query); } - - Ok(false) -} - -fn ui(frame: &mut Frame<'_>, app: &App) { - let chunks = Layout::default() - .direction(Direction::Vertical) - .constraints([ - Constraint::Min(3), - Constraint::Length(3), - Constraint::Length(1), - ]) - .split(frame.area()); - - let output = Paragraph::new(app.output.as_str()) - .block(Block::default().title("Results").borders(Borders::ALL)) - .wrap(Wrap { trim: false }); - frame.render_widget(output, chunks[0]); - - let (input_view, cursor_col) = app.input_view(chunks[1]); - let input = - Paragraph::new(input_view).block(Block::default().title("SQL").borders(Borders::ALL)); - frame.render_widget(input, chunks[1]); - frame.set_cursor_position((chunks[1].x + 1 + cursor_col, chunks[1].y + 1)); - - let status = Paragraph::new(Line::from(app.status.as_str())) - .style(Style::default()) - .block(Block::default().borders(Borders::TOP)) - .wrap(Wrap { trim: true }); - frame.render_widget(status, chunks[2]); } fn format_response(response: &SqlResponse, json_output: bool) -> Result { @@ -823,159 +698,6 @@ fn pad_cell(cell: &str, width: usize) -> String { out } -struct App { - input: String, - cursor: usize, - output: String, - status: String, - history: Vec, - history_index: Option, - json_output: bool, - lint_mode: String, -} - -impl App { - fn new(json_output: bool, lint_mode: String) -> Self { - Self { - input: String::new(), - cursor: 0, - output: String::new(), - status: "Enter SQL and press Enter. Ctrl+C to exit.".to_string(), - history: Vec::new(), - history_index: None, - json_output, - lint_mode, - } - } - - fn insert_char(&mut self, ch: char) { - self.input.insert(self.cursor, ch); - self.cursor += ch.len_utf8(); - self.history_index = None; - } - - fn backspace(&mut self) { - if self.cursor == 0 { - return; - } - let new_cursor = prev_char_boundary(&self.input, self.cursor); - self.input.replace_range(new_cursor..self.cursor, ""); - self.cursor = new_cursor; - self.history_index = None; - } - - fn delete(&mut self) { - if self.cursor >= self.input.len() { - return; - } - let next_cursor = next_char_boundary(&self.input, self.cursor); - self.input.replace_range(self.cursor..next_cursor, ""); - self.history_index = None; - } - - fn move_left(&mut self) { - if self.cursor == 0 { - return; - } - self.cursor = prev_char_boundary(&self.input, self.cursor); - } - - fn move_right(&mut self) { - if self.cursor >= self.input.len() { - return; - } - self.cursor = next_char_boundary(&self.input, self.cursor); - } - - fn move_home(&mut self) { - self.cursor = 0; - } - - fn move_end(&mut self) { - self.cursor = self.input.len(); - } - - fn clear_input(&mut self) { - self.input.clear(); - self.cursor = 0; - self.history_index = None; - } - - fn push_history(&mut self, query: &str) { - if query.trim().is_empty() { - return; - } - if self.history.last().map(String::as_str) != Some(query) { - self.history.push(query.to_string()); - } - self.history_index = None; - } - - fn history_prev(&mut self) { - if self.history.is_empty() { - return; - } - let next_index = match self.history_index { - None => self.history.len().saturating_sub(1), - Some(0) => 0, - Some(idx) => idx - 1, - }; - self.history_index = Some(next_index); - self.input = self.history[next_index].clone(); - self.cursor = self.input.len(); - } - - fn history_next(&mut self) { - let Some(idx) = self.history_index else { - return; - }; - let next_index = idx + 1; - if next_index >= self.history.len() { - self.history_index = None; - self.clear_input(); - return; - } - self.history_index = Some(next_index); - self.input = self.history[next_index].clone(); - self.cursor = self.input.len(); - } - - fn input_view(&self, area: Rect) -> (String, u16) { - let available_width = area.width.saturating_sub(2) as usize; - if available_width == 0 { - return (String::new(), 0); - } - - let mut start = self.cursor.saturating_sub(available_width); - - while start > 0 && !self.input.is_char_boundary(start) { - start -= 1; - } - - let mut end = (start + available_width).min(self.input.len()); - while end < self.input.len() && !self.input.is_char_boundary(end) { - end += 1; - } - - let visible = self.input[start..end].to_string(); - let cursor_col = self.cursor.saturating_sub(start) as u16; - (visible, cursor_col) - } -} - -fn prev_char_boundary(s: &str, idx: usize) -> usize { - s[..idx].char_indices().last().map(|(i, _)| i).unwrap_or(0) -} - -fn next_char_boundary(s: &str, idx: usize) -> usize { - if idx >= s.len() { - return s.len(); - } - let mut iter = s[idx..].char_indices(); - iter.next(); - iter.next().map(|(i, _)| idx + i).unwrap_or_else(|| s.len()) -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/ui/line_editor.rs b/src/ui/line_editor.rs new file mode 100644 index 0000000..0f6a91a --- /dev/null +++ b/src/ui/line_editor.rs @@ -0,0 +1,325 @@ +use std::io::{self, IsTerminal, Write}; + +use anyhow::{Context, Result}; +use crossterm::{ + cursor::MoveToColumn, + event::{ + read as read_terminal_event, Event as TerminalEvent, KeyCode, KeyEvent, KeyEventKind, + KeyModifiers, + }, + queue, + terminal::{disable_raw_mode, enable_raw_mode, Clear, ClearType}, +}; +use unicode_width::UnicodeWidthStr; + +struct LineEditor { + input: String, + cursor: usize, + history: Vec, + history_index: Option, + draft: Option, +} + +impl LineEditor { + fn new(history: Vec) -> Self { + Self { + input: String::new(), + cursor: 0, + history, + history_index: None, + draft: None, + } + } + + fn input(&self) -> &str { + &self.input + } + + fn cursor_prefix(&self) -> &str { + &self.input[..self.cursor] + } + + fn clear_input(&mut self) { + self.input.clear(); + self.cursor = 0; + self.history_index = None; + self.draft = None; + } + + fn add_history(&mut self, value: &str) { + let value = value.trim(); + if value.is_empty() { + return; + } + if self.history.last().map(String::as_str) != Some(value) { + self.history.push(value.to_string()); + } + self.history_index = None; + self.draft = None; + } + + fn handle_key(&mut self, key: KeyEvent) -> bool { + match key.code { + KeyCode::Backspace => self.backspace(), + KeyCode::Delete => self.delete(), + KeyCode::Left => self.move_left(), + KeyCode::Right => self.move_right(), + KeyCode::Home => self.move_home(), + KeyCode::End => self.move_end(), + KeyCode::Up => self.history_prev(), + KeyCode::Down => self.history_next(), + KeyCode::Char(ch) + if !key.modifiers.contains(KeyModifiers::CONTROL) + && !key.modifiers.contains(KeyModifiers::ALT) => + { + self.insert_char(ch); + } + _ => return false, + } + true + } + + fn insert_char(&mut self, ch: char) { + self.clear_history_selection(); + self.input.insert(self.cursor, ch); + self.cursor += ch.len_utf8(); + } + + fn backspace(&mut self) { + if self.cursor == 0 { + return; + } + self.clear_history_selection(); + let new_cursor = prev_char_boundary(&self.input, self.cursor); + self.input.replace_range(new_cursor..self.cursor, ""); + self.cursor = new_cursor; + } + + fn delete(&mut self) { + if self.cursor >= self.input.len() { + return; + } + self.clear_history_selection(); + let next_cursor = next_char_boundary(&self.input, self.cursor); + self.input.replace_range(self.cursor..next_cursor, ""); + } + + fn move_left(&mut self) { + if self.cursor == 0 { + return; + } + self.cursor = prev_char_boundary(&self.input, self.cursor); + } + + fn move_right(&mut self) { + if self.cursor >= self.input.len() { + return; + } + self.cursor = next_char_boundary(&self.input, self.cursor); + } + + fn move_home(&mut self) { + self.cursor = 0; + } + + fn move_end(&mut self) { + self.cursor = self.input.len(); + } + + fn history_prev(&mut self) { + if self.history.is_empty() { + return; + } + let next_index = match self.history_index { + None => { + self.draft = Some(self.input.clone()); + self.history.len().saturating_sub(1) + } + Some(0) => 0, + Some(idx) => idx - 1, + }; + self.history_index = Some(next_index); + self.input = self.history[next_index].clone(); + self.cursor = self.input.len(); + } + + fn history_next(&mut self) { + let Some(idx) = self.history_index else { + return; + }; + let next_index = idx + 1; + if next_index >= self.history.len() { + self.history_index = None; + self.input = self.draft.take().unwrap_or_default(); + self.cursor = self.input.len(); + return; + } + self.history_index = Some(next_index); + self.input = self.history[next_index].clone(); + self.cursor = self.input.len(); + } + + fn clear_history_selection(&mut self) { + if self.history_index.take().is_some() { + self.draft = None; + } + } +} + +pub struct LinePrompt { + editor: LineEditor, +} + +impl LinePrompt { + pub fn new(history: Vec) -> Self { + Self { + editor: LineEditor::new(history), + } + } + + pub fn add_history(&mut self, value: &str) { + self.editor.add_history(value); + } + + pub fn read_line(&mut self, prompt: &str, prompt_width: usize) -> Result> { + if !io::stdin().is_terminal() || !io::stdout().is_terminal() { + print!("{prompt}"); + io::stdout().flush()?; + let mut input = String::new(); + if io::stdin().read_line(&mut input)? == 0 { + println!(); + return Ok(None); + } + return Ok(Some(input.trim_end_matches(['\r', '\n']).to_string())); + } + + let _raw_mode = RawModeGuard::enable()?; + render_prompt_line(prompt, prompt_width, &self.editor)?; + + loop { + let TerminalEvent::Key(key) = read_terminal_event()? else { + continue; + }; + if key.kind != KeyEventKind::Press { + continue; + } + match key.code { + KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => { + finish_prompt_line()?; + return Ok(None); + } + KeyCode::Char('d') + if key.modifiers.contains(KeyModifiers::CONTROL) + && self.editor.input().is_empty() => + { + finish_prompt_line()?; + return Ok(None); + } + KeyCode::Char('j') | KeyCode::Char('m') + if key.modifiers.contains(KeyModifiers::CONTROL) => + { + return finish_with_input(&mut self.editor); + } + KeyCode::Enter => return finish_with_input(&mut self.editor), + _ => { + self.editor.handle_key(key); + } + } + render_prompt_line(prompt, prompt_width, &self.editor)?; + } + } +} + +struct RawModeGuard; + +impl RawModeGuard { + fn enable() -> Result { + enable_raw_mode().context("failed to enable terminal raw mode")?; + Ok(Self) + } +} + +impl Drop for RawModeGuard { + fn drop(&mut self) { + let _ = disable_raw_mode(); + } +} + +fn render_prompt_line(prompt: &str, prompt_width: usize, editor: &LineEditor) -> Result<()> { + let cursor_col = prompt_width + UnicodeWidthStr::width(editor.cursor_prefix()); + let mut stdout = io::stdout(); + queue!(stdout, MoveToColumn(0), Clear(ClearType::CurrentLine))?; + write!(stdout, "{}{}", prompt, editor.input())?; + queue!( + stdout, + MoveToColumn(cursor_col.min(u16::MAX as usize) as u16) + )?; + stdout.flush()?; + Ok(()) +} + +fn finish_with_input(editor: &mut LineEditor) -> Result> { + let input = editor.input().to_string(); + editor.clear_input(); + finish_prompt_line()?; + Ok(Some(input)) +} + +fn finish_prompt_line() -> Result<()> { + let mut stdout = io::stdout(); + write!(stdout, "\r\n")?; + stdout.flush()?; + Ok(()) +} + +fn prev_char_boundary(s: &str, idx: usize) -> usize { + s[..idx].char_indices().last().map(|(i, _)| i).unwrap_or(0) +} + +fn next_char_boundary(s: &str, idx: usize) -> usize { + if idx >= s.len() { + return s.len(); + } + let mut iter = s[idx..].char_indices(); + iter.next(); + iter.next().map(|(i, _)| idx + i).unwrap_or_else(|| s.len()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn line_editor_history_preserves_draft() { + let mut editor = LineEditor::new(vec!["first".to_string(), "second".to_string()]); + editor.insert_char('d'); + editor.insert_char('r'); + editor.insert_char('a'); + editor.insert_char('f'); + editor.insert_char('t'); + + editor.history_prev(); + assert_eq!(editor.input(), "second"); + + editor.history_prev(); + assert_eq!(editor.input(), "first"); + + editor.history_next(); + assert_eq!(editor.input(), "second"); + + editor.history_next(); + assert_eq!(editor.input(), "draft"); + } + + #[test] + fn line_editor_handles_utf8_cursor_movement() { + let mut editor = LineEditor::new(Vec::new()); + editor.insert_char('a'); + editor.insert_char('é'); + editor.insert_char('b'); + editor.move_left(); + editor.backspace(); + + assert_eq!(editor.input(), "ab"); + } +} diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 3f51e64..1bc9923 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -3,6 +3,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use dialoguer::console::Term; +mod line_editor; mod pager; pub mod prompt_render; mod ratatui_table; @@ -51,6 +52,7 @@ pub fn is_interactive() -> bool { std::io::stdin().is_terminal() && !NO_INPUT.load(Ordering::Relaxed) } +pub use line_editor::LinePrompt; pub use pager::print_with_pager; pub use ratatui_table::{ box_with_title, render_experiment_summary_table, summary_metric_unit, SummaryExperimentColumn,