diff --git a/src/topics/api.rs b/src/topics/api.rs index 736276ee..350d25dd 100644 --- a/src/topics/api.rs +++ b/src/topics/api.rs @@ -1,6 +1,7 @@ -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; -use anyhow::{bail, Result}; +use anyhow::{bail, Context, Result}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -19,6 +20,10 @@ const DEFAULT_TOPIC_IDLE_SECONDS: i64 = 10 * 60; const DEFAULT_TOPIC_SAMPLING_RATE: f64 = 1.0; const DEFAULT_TOPIC_EMBEDDING_MODEL: &str = "brain-embedding-1"; const MAX_STATUS_PROGRESS_WINDOW_SECONDS: i64 = 24 * 60 * 60; +const ORG_USERS_PAGE_LIMIT: usize = 1000; +const TOPIC_REPO_ROOT_SPAN_PAGE_LIMIT: usize = 1000; +const TOPIC_REPO_ROOT_SPAN_MAX_IDS: usize = 5000; +const TOPIC_TRACE_CURSOR_PREFIX: &str = "bt-topic-traces-v1:"; #[derive(Debug, Clone, Serialize)] pub struct TopicsStatusReport { @@ -56,6 +61,234 @@ pub struct TopicMapConfigUpdate { pub topic_map_id: String, } +#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, clap::ValueEnum)] +#[serde(rename_all = "snake_case")] +pub enum TopicExploreSort { + Count, + Tokens, + Cost, + AvgTokens, + AvgCost, + Recent, +} + +impl Default for TopicExploreSort { + fn default() -> Self { + Self::Count + } +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, clap::ValueEnum)] +#[serde(rename_all = "snake_case")] +pub enum TopicTraceSort { + Recent, + Tokens, + Cost, +} + +impl Default for TopicTraceSort { + fn default() -> Self { + Self::Recent + } +} + +impl From for TopicExploreSort { + fn from(sort: TopicTraceSort) -> Self { + match sort { + TopicTraceSort::Recent => Self::Recent, + TopicTraceSort::Tokens => Self::Tokens, + TopicTraceSort::Cost => Self::Cost, + } + } +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicsExploreFacetsReport { + pub project: TopicsProjectSummary, + pub facets: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicClassificationsReport { + pub project: TopicsProjectSummary, + pub topic_map: TopicExploreTopicMap, + pub classifications: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicTracesReport { + pub project: TopicsProjectSummary, + pub topic_map: TopicExploreTopicMap, + pub topic: TopicTraceSelection, + pub traces: Vec, + pub next_cursor: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicExploreFacet { + pub automation_id: String, + pub automation_name: String, + pub facet: Option, + pub topic_map: String, + pub topic_map_id: String, + pub version: Option, + pub eligible: usize, + pub labeled: usize, + pub processing: usize, + pub errors: usize, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicExploreTopicMap { + pub automation_id: String, + pub automation_name: String, + pub facet: Option, + pub topic_map: String, + pub topic_map_id: String, + pub version: Option, + pub classification_path: String, + pub btql_filter: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicClassificationRow { + pub topic: String, + pub topic_id: String, + pub traces: usize, + pub tokens: f64, + pub cost: f64, + pub avg_tokens: f64, + pub avg_cost: f64, + pub latest: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicTraceSelection { + pub topic: Option, + pub topic_id: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopicTraceRow { + pub created: Option, + pub root_span_id: String, + pub span_id: Option, + pub row_id: Option, + pub created_by_user_id: Option, + pub created_by_user_name: Option, + pub created_by_user_email: Option, + pub git_origin_url: Option, + pub repo: Option, + pub topic: Option, + pub topic_id: Option, + pub tokens: f64, + pub cost: f64, + pub duration_seconds: Option, + pub input: Option, + pub app_url: String, + #[serde(skip_serializing)] + pagination_key: Option, + #[serde(skip_serializing)] + sort_value: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +struct TopicTracePaginationCursor { + version: u8, + sort: TopicExploreSort, + sort_value: TopicTraceCursorValue, + pagination_key: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(tag = "type", content = "value", rename_all = "snake_case")] +enum TopicTraceCursorValue { + Number(f64), + String(String), +} + +#[derive(Debug, Clone, Deserialize)] +struct OrgUser { + id: String, + #[serde(default)] + given_name: Option, + #[serde(default)] + family_name: Option, + #[serde(default)] + email: Option, +} + +#[derive(Debug, Deserialize)] +struct OrgUsersListResponse { + objects: Vec, +} + +#[derive(Debug, Default)] +pub(super) struct OrgUsersCache { + users_by_id: Option>, +} + +impl OrgUser { + fn display_name(&self) -> Option { + let given = self.given_name.as_deref().unwrap_or_default().trim(); + let family = self.family_name.as_deref().unwrap_or_default().trim(); + let name = match (given.is_empty(), family.is_empty()) { + (true, true) => None, + (false, true) => Some(given.to_string()), + (true, false) => Some(family.to_string()), + (false, false) => Some(format!("{given} {family}")), + }; + name.or_else(|| { + self.email + .as_deref() + .map(str::trim) + .filter(|email| !email.is_empty()) + .map(ToString::to_string) + }) + } +} + +impl OrgUsersCache { + async fn hydrate_trace_users( + &mut self, + client: &ApiClient, + traces: &mut [TopicTraceRow], + ) -> Result<()> { + let user_ids = traces + .iter() + .filter_map(|trace| trace.created_by_user_id.as_deref()) + .collect::>(); + if user_ids.is_empty() { + return Ok(()); + } + + let users = self.users_by_id(client).await?; + for trace in traces { + let Some(user_id) = trace.created_by_user_id.as_deref() else { + continue; + }; + let Some(user) = users.get(user_id) else { + continue; + }; + trace.created_by_user_name = user.display_name(); + trace.created_by_user_email = user.email.clone(); + } + + Ok(()) + } + + async fn users_by_id(&mut self, client: &ApiClient) -> Result<&HashMap> { + if self.users_by_id.is_none() { + self.users_by_id = Some(fetch_org_users(client).await?); + } + + Ok(self + .users_by_id + .as_ref() + .expect("org users cache initialized")) + } +} + #[derive(Debug, Clone, Serialize)] struct TopicMapReportUrlRequest<'a> { function_id: &'a str, @@ -399,6 +632,283 @@ pub async fn fetch_topics_status( }) } +pub async fn topic_explore_filter_clause( + ctx: &ProjectContext, + since: Option<&str>, + window: &str, + extra_filter: Option<&str>, + repo: Option<&str>, + print_queries: bool, +) -> Result { + let time_clause = topic_explore_time_filter_clause(since, window)?; + let repo_filter = + topic_repo_root_span_filter_clause(ctx, repo, &time_clause, print_queries).await?; + + Ok(combine_filter_clauses([ + Some(time_clause), + repo_filter, + extra_filter + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string), + ])) +} + +fn topic_explore_time_filter_clause(since: Option<&str>, window: &str) -> Result { + Ok( + if let Some(ts) = since.map(str::trim).filter(|value| !value.is_empty()) { + format!("created >= {}", btql_string_literal(ts)) + } else { + let seconds = crate::utils::parse_duration_to_seconds(window)?; + if seconds == 0 { + bail!("--window must be greater than zero"); + } + format!("created >= NOW() - INTERVAL {seconds} SECOND") + }, + ) +} + +pub async fn fetch_topics_explore_facets( + ctx: &ProjectContext, + automation_id: Option<&str>, + base_filter_clause: &str, + print_queries: bool, +) -> Result { + let rows = list_topic_automation_rows(&ctx.client, &ctx.project.id).await?; + let rows = filter_or_resolve_topic_automation_rows(rows, automation_id)?; + let mut function_cache = HashMap::new(); + let mut facets = Vec::new(); + + for row in &rows { + let config = row + .get("config") + .and_then(Value::as_object) + .cloned() + .unwrap_or_default(); + let automation_id = stringish_value(row.get("id")).unwrap_or_default(); + let automation_name = string_value(row.get("name")).unwrap_or_else(|| "Topics".to_string()); + let automation_filter = combine_filter_clauses([ + Some(base_filter_clause.to_string()), + string_value(config.get("btql_filter")), + ]); + + let cursor = fetch_cursor_snapshot(&ctx.client, &ctx.project.id, &automation_id).await?; + let topic_bars = build_topic_status_bars(&ctx.client, &mut function_cache, &config).await?; + let facet_bars = + build_facet_status_bars(&ctx.client, &mut function_cache, &config, &topic_bars).await?; + let progress = fetch_topic_automation_progress( + &ctx.client, + &ctx.project.id, + &automation_filter, + &cursor, + &facet_bars, + &topic_bars, + print_queries, + ) + .await?; + let topic_progress_by_name = progress + .topics + .into_iter() + .map(|item| (item.name.clone(), item)) + .collect::>(); + let topic_maps = summarize_topic_map_functions( + &ctx.client, + &mut function_cache, + config.get("topic_map_functions"), + ) + .await?; + + for topic_map in topic_maps { + let counts = topic_progress_by_name.get(&topic_map.name); + facets.push(TopicExploreFacet { + automation_id: automation_id.clone(), + automation_name: automation_name.clone(), + facet: topic_map.source_facet.clone(), + topic_map: topic_map.name.clone(), + topic_map_id: topic_map.id.clone().unwrap_or_default(), + version: topic_map.version.clone(), + eligible: counts.map(|item| item.matched_count).unwrap_or(0), + labeled: counts.map(|item| item.completed_count).unwrap_or(0), + processing: counts.map(|item| item.processing_count).unwrap_or(0), + errors: counts.map(|item| item.error_count).unwrap_or(0), + }); + } + } + + facets.sort_by(|left, right| { + left.facet + .cmp(&right.facet) + .then(left.topic_map.cmp(&right.topic_map)) + .then(left.topic_map_id.cmp(&right.topic_map_id)) + }); + + Ok(TopicsExploreFacetsReport { + project: topics_project_summary(ctx), + facets, + }) +} + +pub async fn fetch_topic_classifications( + ctx: &ProjectContext, + automation_id: Option<&str>, + facet: Option<&str>, + topic_map: Option<&str>, + sort: TopicExploreSort, + limit: usize, + base_filter_clause: &str, + print_queries: bool, +) -> Result { + if limit == 0 { + bail!("--limit must be greater than 0"); + } + let topic_map = resolve_topic_explore_topic_map(ctx, automation_id, facet, topic_map).await?; + let filter_clause = topic_map_filter_clause(&topic_map, base_filter_clause); + let query = build_topic_classifications_query( + &ctx.project.id, + &topic_map.topic_map, + &topic_map.topic_map_id, + &filter_clause, + sort, + limit, + ); + maybe_print_topic_query(print_queries, "classifications", &query); + let response = execute_btql_value(&ctx.client, &query).await?; + let classifications = btql_data_rows(&response) + .into_iter() + .map(topic_classification_row_from_btql) + .collect(); + + Ok(TopicClassificationsReport { + project: topics_project_summary(ctx), + topic_map, + classifications, + }) +} + +pub async fn fetch_topic_traces( + ctx: &ProjectContext, + automation_id: Option<&str>, + facet: Option<&str>, + topic_map: Option<&str>, + topic: Option<&str>, + topic_id: Option<&str>, + sort: TopicExploreSort, + limit: usize, + cursor: Option<&str>, + base_filter_clause: &str, + print_queries: bool, +) -> Result { + let mut users_cache = OrgUsersCache::default(); + fetch_topic_traces_with_user_cache( + ctx, + automation_id, + facet, + topic_map, + topic, + topic_id, + sort, + limit, + cursor, + base_filter_clause, + print_queries, + &mut users_cache, + ) + .await +} + +pub(super) async fn fetch_topic_traces_with_user_cache( + ctx: &ProjectContext, + automation_id: Option<&str>, + facet: Option<&str>, + topic_map: Option<&str>, + topic: Option<&str>, + topic_id: Option<&str>, + sort: TopicExploreSort, + limit: usize, + cursor: Option<&str>, + base_filter_clause: &str, + print_queries: bool, + users_cache: &mut OrgUsersCache, +) -> Result { + if limit == 0 { + bail!("--limit must be greater than 0"); + } + if topic.is_none() && topic_id.is_none() { + bail!("topic label selection required; pass --topic-id or --topic after choosing a row from `bt topics classifications`"); + } + + let topic_map = resolve_topic_explore_topic_map(ctx, automation_id, facet, topic_map).await?; + let topic_cursor = parse_topic_trace_cursor(cursor, sort)?; + let backend_cursor = if topic_cursor.is_some() { + None + } else { + cursor.filter(|cursor| !cursor.trim().is_empty()) + }; + let mut filter_clause = topic_map_filter_clause(&topic_map, base_filter_clause); + filter_clause = combine_filter_clauses([ + Some(filter_clause), + topic_filter_clause(&topic_map, topic, topic_id)?, + topic_cursor + .as_ref() + .map(|cursor| topic_trace_cursor_filter_clause(sort, cursor)) + .transpose()?, + ]); + let fetch_limit = if backend_cursor.is_some() { + limit + } else { + limit.saturating_add(1) + }; + let query = build_topic_traces_query( + &ctx.project.id, + &topic_map.topic_map, + &topic_map.topic_map_id, + &filter_clause, + sort, + fetch_limit, + backend_cursor, + ); + maybe_print_topic_query(print_queries, "traces", &query); + let response = execute_btql_value(&ctx.client, &query).await?; + let returned_rows = btql_data_len(&response); + let project_url = app_project_url( + &ctx.app_url, + ctx.client.org_name(), + &ctx.project.name, + &["logs"], + ); + let mut traces = btql_data_rows(&response) + .into_iter() + .map(|row| topic_trace_row_from_btql(row, &project_url)) + .collect::>(); + let next_cursor = topic_trace_next_cursor(&traces, sort, limit)? + .or_else(|| next_cursor_if_full_page(btql_cursor(&response), returned_rows, limit)); + if traces.len() > limit { + traces.truncate(limit); + } + if let Err(err) = + hydrate_trace_root_metadata(&ctx.client, &ctx.project.id, &mut traces, print_queries).await + { + eprintln!("warning: failed to resolve trace root metadata: {err}"); + } + if let Err(err) = users_cache + .hydrate_trace_users(&ctx.client, &mut traces) + .await + { + eprintln!("warning: failed to resolve trace users: {err}"); + } + + Ok(TopicTracesReport { + project: topics_project_summary(ctx), + topic_map, + topic: TopicTraceSelection { + topic: topic.map(ToString::to_string), + topic_id: topic_id.map(ToString::to_string), + }, + traces, + next_cursor, + }) +} + pub async fn poke_topic_automations(ctx: &ProjectContext) -> Result { let rows = list_topic_automation_rows(&ctx.client, &ctx.project.id).await?; let mut queued = Vec::with_capacity(rows.len()); @@ -1273,6 +1783,7 @@ async fn build_topic_automation_status( &cursor, &facet_bars, &topic_bars, + false, ) .await?; total_traces = progress.total_traces; @@ -1835,83 +2346,1037 @@ fn ensure_facet_bar( } } -async fn fetch_topic_automation_progress( - client: &ApiClient, - project_id: &str, - time_filter_clause: &str, - cursor_status: &AutomationCursorSnapshot, - facet_bars: &[FacetStatusBar], - topic_bars: &[TopicStatusBar], -) -> Result { - let pending_min_executed_xact_id = cursor_status.pending_min_executed_xact_id.as_deref(); - let mut measure_expressions = Vec::::new(); +async fn resolve_topic_explore_topic_map( + ctx: &ProjectContext, + automation_id: Option<&str>, + facet: Option<&str>, + topic_map: Option<&str>, +) -> Result { + let topic_maps = list_topic_explore_topic_maps(ctx, automation_id).await?; + if topic_maps.is_empty() { + bail!("no configured topic maps found; run `bt topics config` to inspect Topics setup"); + } - let facet_paths = facet_bars - .iter() - .map(|bar| bar.facet_path.as_str()) + let matches = topic_maps + .into_iter() + .filter(|candidate| { + facet + .map(|facet| { + candidate + .facet + .as_deref() + .map(|candidate| selector_matches(candidate, facet)) + .unwrap_or(false) + }) + .unwrap_or(true) + }) + .filter(|candidate| { + topic_map + .map(|topic_map| { + selector_matches(&candidate.topic_map, topic_map) + || selector_matches(&candidate.topic_map_id, topic_map) + }) + .unwrap_or(true) + }) .collect::>(); - if !facet_paths.is_empty() { - let facet_coverage_predicate = facet_paths - .iter() - .map(|facet_path| format!("({facet_path} != 'no_match')")) - .collect::>() - .join(" OR "); - measure_expressions.push(format!( - "count(({facet_coverage_predicate}) ? 1 : null) as facet_current_marked_traces" - )); - } - for (index, bar) in facet_bars.iter().enumerate() { - let prefix = format!("facet_{index}"); - measure_expressions.push(format!( - "count((({} != 'no_match')) ? 1 : null) as {prefix}_current_marked_traces", - bar.facet_path - )); - measure_expressions.push(format!( - "count({}) as {prefix}_current_completed_output_traces", - bar.facet_path - )); - if bar.function_keys.is_empty() { - continue; + match matches.len() { + 0 => bail!( + "topic map selection did not match any configured topic map; run `bt topics facets` to list available facets and topic maps" + ), + 1 => Ok(matches.into_iter().next().expect("single topic map match")), + _ => { + let choices = matches + .iter() + .take(5) + .map(format_topic_map_choice) + .collect::>() + .join(", "); + let suffix = if matches.len() > 5 { ", ..." } else { "" }; + bail!( + "topic map selection matched multiple entries ({choices}{suffix}); re-run with --facet or --topic-map" + ) } + } +} - let mut completed_predicates = Vec::new(); - let mut inflight_predicates = Vec::new(); - let mut error_predicates = Vec::new(); - for function_key in &bar.function_keys { - let predicates = - build_triggered_function_predicates(function_key, pending_min_executed_xact_id); - completed_predicates.push(format!("({})", predicates.completed_predicate)); - inflight_predicates.push(format!("({})", predicates.inflight_predicate)); - error_predicates.push(format!("({})", predicates.error_predicate)); - } +async fn list_topic_explore_topic_maps( + ctx: &ProjectContext, + automation_id: Option<&str>, +) -> Result> { + let rows = list_topic_automation_rows(&ctx.client, &ctx.project.id).await?; + let rows = filter_or_resolve_topic_automation_rows(rows, automation_id)?; + let mut function_cache = HashMap::new(); + let mut topic_maps = Vec::new(); - measure_expressions.push(format!( - "count(({} ) ? 1 : null) as {prefix}_current_completed_traces", - completed_predicates.join(" OR ") - )); - measure_expressions.push(format!( - "count(({} ) ? 1 : null) as {prefix}_current_inflight_traces", - inflight_predicates.join(" OR ") - )); - measure_expressions.push(format!( - "count(({} ) ? 1 : null) as {prefix}_current_error_traces", - error_predicates.join(" OR ") - )); + for row in &rows { + let automation = + build_topic_automation_config(&ctx.client, row, &mut function_cache).await?; + for topic_map in &automation.topic_map_functions { + let Some(topic_map_id) = topic_map.id.clone() else { + continue; + }; + let combined_filter = combine_optional_filter_clauses([ + automation.btql_filter.clone(), + topic_map.btql_filter.clone(), + ]); + topic_maps.push(TopicExploreTopicMap { + automation_id: automation.id.clone(), + automation_name: automation.name.clone(), + facet: topic_map.source_facet.clone(), + topic_map: topic_map.name.clone(), + topic_map_id, + version: topic_map.version.clone(), + classification_path: escape_btql_ident_path(&[ + "classifications", + topic_map.name.as_str(), + ]), + btql_filter: combined_filter, + }); + } } - for (index, bar) in topic_bars.iter().enumerate() { - let prefix = format!("topic_{index}"); - measure_expressions.push(format!( - "count(({}) ? 1 : null) as {prefix}_current_eligible_traces", - bar.eligible_predicate - )); - measure_expressions.push(format!( - "count({}) as {prefix}_current_labeled_traces", - bar.classification_path - )); - if let Some(function_key) = bar.function_key.as_deref() { - let predicates = + topic_maps.sort_by(|left, right| { + left.facet + .cmp(&right.facet) + .then(left.topic_map.cmp(&right.topic_map)) + .then(left.topic_map_id.cmp(&right.topic_map_id)) + }); + Ok(topic_maps) +} + +fn selector_matches(candidate: &str, selector: &str) -> bool { + candidate == selector || candidate.eq_ignore_ascii_case(selector) +} + +fn format_topic_map_choice(topic_map: &TopicExploreTopicMap) -> String { + format!( + "{} / {} (topic map id: {}, automation: {} [{}])", + topic_map.facet.as_deref().unwrap_or("Ungrouped"), + topic_map.topic_map, + topic_map.topic_map_id, + topic_map.automation_name, + topic_map.automation_id + ) +} + +fn topic_map_filter_clause(topic_map: &TopicExploreTopicMap, base_filter_clause: &str) -> String { + let source_type_path = escape_btql_ident_path(&[ + "classifications", + topic_map.topic_map.as_str(), + "source", + "type", + ]); + let source_id_path = escape_btql_ident_path(&[ + "classifications", + topic_map.topic_map.as_str(), + "source", + "id", + ]); + combine_filter_clauses([ + Some(base_filter_clause.to_string()), + topic_map.btql_filter.clone(), + Some(format!("{} IS NOT NULL", topic_map.classification_path)), + Some(format!("{source_type_path} = 'function'")), + Some(format!( + "{source_id_path} = {}", + btql_string_literal(&topic_map.topic_map_id) + )), + ]) +} + +fn topic_filter_clause( + topic_map: &TopicExploreTopicMap, + topic: Option<&str>, + topic_id: Option<&str>, +) -> Result> { + match (topic, topic_id) { + (Some(_), Some(_)) => bail!("use either --topic-id or --topic, not both"), + (None, None) => Ok(None), + (None, Some(topic_id)) => { + let id_path = + escape_btql_ident_path(&["classifications", topic_map.topic_map.as_str(), "id"]); + Ok(Some(format!( + "{id_path} = {}", + btql_string_literal(topic_id) + ))) + } + (Some(topic), None) => { + let id_path = + escape_btql_ident_path(&["classifications", topic_map.topic_map.as_str(), "id"]); + let label_path = + escape_btql_ident_path(&["classifications", topic_map.topic_map.as_str(), "label"]); + Ok(Some(format!( + "COALESCE({label_path}, {id_path}) = {}", + btql_string_literal(topic) + ))) + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct RepoOrigin { + host: String, + owner_path: String, + repo: String, +} + +impl RepoOrigin { + fn canonical_slug(&self) -> String { + format!("{}/{}/{}", self.host, self.owner_path, self.repo) + } +} + +async fn topic_repo_root_span_filter_clause( + ctx: &ProjectContext, + repo: Option<&str>, + time_clause: &str, + print_queries: bool, +) -> Result> { + let Some(raw) = repo.map(str::trim).filter(|value| !value.is_empty()) else { + return Ok(None); + }; + let origin = parse_repo_origin_selector(raw)?; + let root_span_ids = fetch_topic_repo_root_span_ids( + &ctx.client, + &ctx.project.id, + time_clause, + &origin, + raw, + print_queries, + ) + .await?; + Ok(Some(root_span_id_filter_clause(&root_span_ids))) +} + +async fn fetch_topic_repo_root_span_ids( + client: &ApiClient, + project_id: &str, + time_clause: &str, + origin: &RepoOrigin, + raw: &str, + print_queries: bool, +) -> Result> { + let origin_filter = topic_repo_origin_filter_clause(origin, Some(raw)); + let mut root_span_ids = BTreeSet::new(); + let mut cursor = None::; + + loop { + if root_span_ids.len() >= TOPIC_REPO_ROOT_SPAN_MAX_IDS { + bail!( + "--repo matched at least {} traces; narrow the search with --window or --since before exploring", + TOPIC_REPO_ROOT_SPAN_MAX_IDS + ); + } + + let remaining = TOPIC_REPO_ROOT_SPAN_MAX_IDS - root_span_ids.len(); + let limit = remaining.min(TOPIC_REPO_ROOT_SPAN_PAGE_LIMIT); + let query = build_topic_repo_root_spans_query( + project_id, + time_clause, + &origin_filter, + limit, + cursor.as_deref(), + ); + maybe_print_topic_query(print_queries, "repo-roots", &query); + let response = execute_btql_value(client, &query).await?; + let returned_rows = btql_data_len(&response); + for row in btql_data_rows(&response) { + if let Some(root_span_id) = value_as_string(row.get("root_span_id")) + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + { + root_span_ids.insert(root_span_id); + } + } + + let next_cursor = next_cursor_if_full_page(btql_cursor(&response), returned_rows, limit); + let Some(next_cursor) = next_cursor else { + break; + }; + cursor = Some(next_cursor); + } + + Ok(root_span_ids.into_iter().collect()) +} + +fn topic_repo_origin_filter_clause(origin: &RepoOrigin, raw: Option<&str>) -> String { + let mut variants = repo_origin_url_variants(origin); + if let Some(raw) = raw + .map(trim_repo_selector) + .filter(|value| !value.is_empty()) + { + variants.insert(raw.to_string()); + } + let values = variants + .into_iter() + .map(|value| btql_string_literal(&value)) + .collect::>() + .join(", "); + let path = escape_btql_ident_path(&["metadata", "git_origin_url"]); + format!("{path} IN [{values}]") +} + +fn build_topic_repo_root_spans_query( + project_id: &str, + time_clause: &str, + origin_filter: &str, + limit: usize, + cursor: Option<&str>, +) -> String { + let cursor_clause = cursor + .filter(|cursor| !cursor.trim().is_empty()) + .map(|cursor| format!(" | cursor: {}", btql_json_string_literal(cursor))) + .unwrap_or_default(); + format!( + "select: root_span_id, metadata.git_origin_url as git_origin_url | from: project_logs({}) spans | filter: ({time_clause}) AND (span_id = root_span_id) AND ({origin_filter}) | preview_length: 1 | sort: created DESC | limit: {limit}{cursor_clause}", + btql_string_literal(project_id), + ) +} + +fn repo_origin_url_variants(origin: &RepoOrigin) -> BTreeSet { + let mut variants = BTreeSet::new(); + for owner_repo_path in repo_owner_repo_path_variants(origin) { + variants.insert(format!("{}/{}.git", origin.host, owner_repo_path)); + variants.insert(format!("{}/{}", origin.host, owner_repo_path)); + variants.insert(format!("{}.git", owner_repo_path)); + variants.insert(owner_repo_path.clone()); + variants.insert(format!("https://{}/{}.git", origin.host, owner_repo_path)); + variants.insert(format!("https://{}/{}", origin.host, owner_repo_path)); + variants.insert(format!("http://{}/{}.git", origin.host, owner_repo_path)); + variants.insert(format!("http://{}/{}", origin.host, owner_repo_path)); + variants.insert(format!("git@{}:{}.git", origin.host, owner_repo_path)); + variants.insert(format!("git@{}:{}", origin.host, owner_repo_path)); + variants.insert(format!("ssh://git@{}/{}.git", origin.host, owner_repo_path)); + variants.insert(format!("ssh://git@{}/{}", origin.host, owner_repo_path)); + } + variants +} + +fn repo_owner_repo_path_variants(origin: &RepoOrigin) -> BTreeSet { + let mut variants = BTreeSet::new(); + variants.insert(format!("{}/{}", origin.owner_path, origin.repo)); + let lower = format!( + "{}/{}", + origin.owner_path.to_ascii_lowercase(), + origin.repo.to_ascii_lowercase() + ); + variants.insert(lower); + variants +} + +fn repo_slug_from_origin_url(value: &str) -> Option { + parse_repo_origin_selector(value) + .ok() + .map(|origin| origin.canonical_slug()) +} + +fn parse_repo_origin_selector(value: &str) -> Result { + let value = trim_repo_selector(value); + if value.is_empty() { + bail!("--repo cannot be empty"); + } + + if let Some(rest) = value.strip_prefix("git@") { + let Some((host, path)) = rest.split_once(':') else { + bail!("invalid --repo git origin; expected git@host:owner/repo"); + }; + return repo_origin_from_host_path(host, path); + } + + if let Some((_, rest)) = value.split_once("://") { + let rest = rest.trim_start_matches('/'); + let rest = rest + .rsplit_once('@') + .map(|(_, without_credentials)| without_credentials) + .unwrap_or(rest); + let Some((authority, path)) = rest.split_once('/') else { + bail!("invalid --repo URL; expected host/owner/repo"); + }; + return repo_origin_from_host_path(authority, path); + } + + if let Some((host, path)) = value.split_once(':') { + if host.contains('.') { + return repo_origin_from_host_path(host, path); + } + } + + let parts = repo_path_parts(value); + match parts.len() { + 2 => repo_origin_from_parts("github.com", &parts), + 3.. => repo_origin_from_parts(parts[0], &parts[1..]), + _ => bail!("invalid --repo; use owner/repo, host/owner/repo, or a full git origin URL"), + } +} + +fn trim_repo_selector(value: &str) -> &str { + value + .trim() + .split(['?', '#']) + .next() + .unwrap_or("") + .trim_end_matches('/') +} + +fn repo_origin_from_host_path(host: &str, path: &str) -> Result { + let host = host.split(':').next().unwrap_or(host); + let parts = repo_path_parts(path); + repo_origin_from_parts(host, &parts) +} + +fn repo_path_parts(value: &str) -> Vec<&str> { + value + .trim_matches('/') + .split('/') + .map(str::trim) + .filter(|part| !part.is_empty()) + .collect() +} + +fn repo_origin_from_parts(host: &str, parts: &[&str]) -> Result { + let host = host.trim().trim_end_matches('/').to_ascii_lowercase(); + if host.is_empty() { + bail!("invalid --repo; repository host cannot be empty"); + } + if parts.len() < 2 { + bail!("invalid --repo; expected owner/repo after the host"); + } + let owner_path = parts[..parts.len() - 1].join("/"); + let repo = parts[parts.len() - 1] + .trim_end_matches('/') + .strip_suffix(".git") + .unwrap_or(parts[parts.len() - 1]) + .to_string(); + if owner_path.trim().is_empty() || repo.trim().is_empty() { + bail!("invalid --repo; expected owner/repo after the host"); + } + + Ok(RepoOrigin { + host, + owner_path, + repo, + }) +} + +fn build_topic_classifications_query( + project_id: &str, + topic_map_name: &str, + _topic_map_id: &str, + filter_clause: &str, + sort: TopicExploreSort, + limit: usize, +) -> String { + let topic_id_path = escape_btql_ident_path(&["classifications", topic_map_name, "id"]); + let topic_label_path = escape_btql_ident_path(&["classifications", topic_map_name, "label"]); + let topic_expr = format!("COALESCE({topic_label_path}, {topic_id_path})"); + format!( + "from: project_logs({}) summary | dimensions: {topic_id_path} as topic_id, {topic_expr} as topic | measures: count_distinct(root_span_id) as traces, sum({}) as tokens, sum({}) as cost, avg({}) as avg_tokens, avg({}) as avg_cost, max(created) as latest | filter: {filter_clause} | sort: {} DESC | limit: {limit}", + btql_string_literal(project_id), + topic_tokens_expr(), + topic_cost_expr(), + topic_tokens_expr(), + topic_cost_expr(), + classification_sort_alias(sort), + ) +} + +fn build_topic_traces_query( + project_id: &str, + topic_map_name: &str, + _topic_map_id: &str, + filter_clause: &str, + sort: TopicExploreSort, + limit: usize, + cursor: Option<&str>, +) -> String { + let topic_id_path = escape_btql_ident_path(&["classifications", topic_map_name, "id"]); + let topic_label_path = escape_btql_ident_path(&["classifications", topic_map_name, "label"]); + let topic_expr = format!("COALESCE({topic_label_path}, {topic_id_path})"); + let sort_expr = trace_sort_expr(sort); + let cursor_clause = cursor + .filter(|cursor| !cursor.trim().is_empty()) + .map(|cursor| format!(" | cursor: {}", btql_json_string_literal(cursor))) + .unwrap_or_default(); + format!( + "select: created, root_span_id, span_id, id, _pagination_key, span_attributes.created_by_user_id as created_by_user_id, metadata.git_origin_url as git_origin_url, {topic_id_path} as topic_id, {topic_expr} as topic, {sort_expr} as sort_value, metrics, input | from: project_logs({}) summary | filter: {filter_clause} | preview_length: 125 | sort: {sort_expr} DESC, _pagination_key DESC | limit: {limit}{cursor_clause}", + btql_string_literal(project_id), + ) +} + +fn build_topic_trace_root_metadata_query(project_id: &str, root_span_ids: &[String]) -> String { + let root_filter = root_span_id_filter_clause(root_span_ids); + format!( + "select: root_span_id, span_attributes.created_by_user_id as created_by_user_id, metadata.git_origin_url as git_origin_url | from: project_logs({}) spans | filter: ({root_filter}) AND (span_id = root_span_id) | preview_length: 1 | limit: {}", + btql_string_literal(project_id), + root_span_ids.len().max(1), + ) +} + +fn root_span_id_filter_clause(root_span_ids: &[String]) -> String { + match root_span_ids { + [] => "root_span_id = ''".to_string(), + [single] => format!("root_span_id = {}", btql_string_literal(single)), + _ => { + let ids = root_span_ids + .iter() + .map(|root_span_id| btql_string_literal(root_span_id)) + .collect::>() + .join(", "); + format!("root_span_id IN [{ids}]") + } + } +} + +fn classification_sort_alias(sort: TopicExploreSort) -> &'static str { + match sort { + TopicExploreSort::Count => "traces", + TopicExploreSort::Tokens => "tokens", + TopicExploreSort::Cost => "cost", + TopicExploreSort::AvgTokens => "avg_tokens", + TopicExploreSort::AvgCost => "avg_cost", + TopicExploreSort::Recent => "latest", + } +} + +fn normalize_trace_sort(sort: TopicExploreSort) -> TopicExploreSort { + match sort { + TopicExploreSort::AvgTokens => TopicExploreSort::Tokens, + TopicExploreSort::AvgCost => TopicExploreSort::Cost, + _ => sort, + } +} + +fn trace_sort_expr(sort: TopicExploreSort) -> &'static str { + let sort = normalize_trace_sort(sort); + match sort { + TopicExploreSort::Tokens => topic_tokens_expr(), + TopicExploreSort::Cost => topic_cost_expr(), + TopicExploreSort::Count | TopicExploreSort::Recent => "created", + TopicExploreSort::AvgTokens | TopicExploreSort::AvgCost => { + unreachable!("normalized above") + } + } +} + +fn topic_explore_sort_name(sort: TopicExploreSort) -> &'static str { + match sort { + TopicExploreSort::Count => "count", + TopicExploreSort::Tokens => "tokens", + TopicExploreSort::Cost => "cost", + TopicExploreSort::AvgTokens => "avg-tokens", + TopicExploreSort::AvgCost => "avg-cost", + TopicExploreSort::Recent => "recent", + } +} + +fn topic_tokens_expr() -> &'static str { + "COALESCE(metrics.total_tokens, metrics.tokens, metrics.prompt_tokens + metrics.completion_tokens, metrics.input_tokens + metrics.output_tokens, 0)" +} + +fn topic_cost_expr() -> &'static str { + "COALESCE(metrics.estimated_cost, metrics.cost, 0)" +} + +fn topics_project_summary(ctx: &ProjectContext) -> TopicsProjectSummary { + TopicsProjectSummary { + id: ctx.project.id.clone(), + name: ctx.project.name.clone(), + org_name: ctx.client.org_name().to_string(), + topics_url: topics_url(&ctx.app_url, ctx.client.org_name(), &ctx.project.name), + } +} + +fn btql_data_rows(response: &Value) -> Vec<&serde_json::Map> { + response + .get("data") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_object) + .collect() +} + +fn btql_data_len(response: &Value) -> usize { + response + .get("data") + .and_then(Value::as_array) + .map(Vec::len) + .unwrap_or(0) +} + +fn btql_cursor(response: &Value) -> Option { + value_as_string(response.get("cursor")).filter(|cursor| !cursor.is_empty()) +} + +fn next_cursor_if_full_page( + cursor: Option, + returned_rows: usize, + requested_limit: usize, +) -> Option { + cursor.filter(|_| requested_limit > 0 && returned_rows >= requested_limit) +} + +fn topic_classification_row_from_btql( + row: &serde_json::Map, +) -> TopicClassificationRow { + TopicClassificationRow { + topic: value_as_string(row.get("topic")).unwrap_or_else(|| "".to_string()), + topic_id: value_as_string(row.get("topic_id")).unwrap_or_default(), + traces: read_btql_count_metric(Some(row), "traces"), + tokens: read_btql_f64_metric(row, "tokens"), + cost: read_btql_f64_metric(row, "cost"), + avg_tokens: read_btql_f64_metric(row, "avg_tokens"), + avg_cost: read_btql_f64_metric(row, "avg_cost"), + latest: value_as_string(row.get("latest")), + } +} + +fn topic_trace_row_from_btql( + row: &serde_json::Map, + project_url: &str, +) -> TopicTraceRow { + let root_span_id = value_as_string(row.get("root_span_id")).unwrap_or_default(); + let span_id = value_as_string(row.get("span_id")).filter(|value| !value.is_empty()); + let mut app_url = format!("{project_url}?r={}", encode(&root_span_id)); + if let Some(span_id) = span_id.as_deref() { + app_url.push_str("&s="); + app_url.push_str(&encode(span_id)); + } + let git_origin_url = trace_git_origin_url(row); + let repo = git_origin_url + .as_deref() + .and_then(repo_slug_from_origin_url); + + TopicTraceRow { + created: value_as_string(row.get("created")), + root_span_id, + span_id, + row_id: value_as_string(row.get("id")), + created_by_user_id: trace_created_by_user_id(row), + created_by_user_name: None, + created_by_user_email: None, + git_origin_url, + repo, + topic: value_as_string(row.get("topic")), + topic_id: value_as_string(row.get("topic_id")), + tokens: metrics_total_tokens(row.get("metrics")).unwrap_or(0.0), + cost: metrics_cost(row.get("metrics")).unwrap_or(0.0), + duration_seconds: metrics_duration_seconds(row.get("metrics")), + input: row.get("input").map(format_preview_value), + app_url, + pagination_key: value_as_string(row.get("_pagination_key")), + sort_value: topic_trace_cursor_value_from_btql(row.get("sort_value")), + } +} + +fn topic_trace_cursor_value_from_btql(value: Option<&Value>) -> Option { + match value { + Some(Value::Number(number)) => number + .as_f64() + .filter(|value| value.is_finite()) + .map(TopicTraceCursorValue::Number), + Some(Value::String(value)) => value + .parse::() + .ok() + .filter(|value| value.is_finite()) + .map(TopicTraceCursorValue::Number) + .or_else(|| Some(TopicTraceCursorValue::String(value.clone()))), + Some(Value::Bool(value)) => Some(TopicTraceCursorValue::String(value.to_string())), + _ => None, + } +} + +fn topic_trace_next_cursor( + traces: &[TopicTraceRow], + sort: TopicExploreSort, + requested_limit: usize, +) -> Result> { + if requested_limit == 0 || traces.len() <= requested_limit { + return Ok(None); + } + + let Some(last_visible) = traces.get(requested_limit.saturating_sub(1)) else { + return Ok(None); + }; + let Some(sort_value) = last_visible.sort_value.clone() else { + return Ok(None); + }; + let Some(pagination_key) = last_visible + .pagination_key + .as_deref() + .filter(|value| !value.trim().is_empty()) + else { + return Ok(None); + }; + + encode_topic_trace_cursor(&TopicTracePaginationCursor { + version: 1, + sort, + sort_value, + pagination_key: pagination_key.to_string(), + }) + .map(Some) +} + +fn encode_topic_trace_cursor(cursor: &TopicTracePaginationCursor) -> Result { + let payload = serde_json::to_vec(cursor)?; + Ok(format!( + "{TOPIC_TRACE_CURSOR_PREFIX}{}", + URL_SAFE_NO_PAD.encode(payload) + )) +} + +fn parse_topic_trace_cursor( + cursor: Option<&str>, + expected_sort: TopicExploreSort, +) -> Result> { + let Some(cursor) = cursor.map(str::trim).filter(|cursor| !cursor.is_empty()) else { + return Ok(None); + }; + let Some(encoded) = cursor.strip_prefix(TOPIC_TRACE_CURSOR_PREFIX) else { + return Ok(None); + }; + + let payload = URL_SAFE_NO_PAD + .decode(encoded) + .context("failed to decode topics trace cursor")?; + let decoded: TopicTracePaginationCursor = + serde_json::from_slice(&payload).context("failed to parse topics trace cursor")?; + if decoded.version != 1 { + bail!( + "unsupported topics trace cursor version {}", + decoded.version + ); + } + if decoded.sort != expected_sort { + bail!( + "cursor was created with --sort {}; this request uses --sort {}", + topic_explore_sort_name(decoded.sort), + topic_explore_sort_name(expected_sort) + ); + } + if decoded.pagination_key.trim().is_empty() { + bail!("topics trace cursor is missing its pagination key"); + } + + Ok(Some(decoded)) +} + +fn topic_trace_cursor_filter_clause( + sort: TopicExploreSort, + cursor: &TopicTracePaginationCursor, +) -> Result { + let sort_expr = trace_sort_expr(sort); + let pagination_key = btql_string_literal(&cursor.pagination_key); + let sort_value = match &cursor.sort_value { + TopicTraceCursorValue::Number(value) => { + if !value.is_finite() { + bail!("topics trace cursor has a non-finite sort value"); + } + value.to_string() + } + TopicTraceCursorValue::String(value) => btql_string_literal(value), + }; + + Ok(format!( + "({sort_expr} < {sort_value}) OR (({sort_expr} = {sort_value}) AND (_pagination_key < {pagination_key}))" + )) +} + +fn trace_created_by_user_id(row: &serde_json::Map) -> Option { + value_as_string(row.get("created_by_user_id")) + .or_else(|| value_as_string(row.get("span_attributes.created_by_user_id"))) + .or_else(|| nested_value_as_string(row, &["span_attributes", "created_by_user_id"])) + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +fn trace_git_origin_url(row: &serde_json::Map) -> Option { + value_as_string(row.get("git_origin_url")) + .or_else(|| value_as_string(row.get("metadata.git_origin_url"))) + .or_else(|| nested_value_as_string(row, &["metadata", "git_origin_url"])) + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +async fn hydrate_trace_root_metadata( + client: &ApiClient, + project_id: &str, + traces: &mut [TopicTraceRow], + print_queries: bool, +) -> Result<()> { + let root_span_ids = traces + .iter() + .filter(|trace| trace.created_by_user_id.is_none() || trace.git_origin_url.is_none()) + .filter_map(|trace| { + let root_span_id = trace.root_span_id.trim(); + (!root_span_id.is_empty()).then(|| root_span_id.to_string()) + }) + .collect::>() + .into_iter() + .collect::>(); + if root_span_ids.is_empty() { + return Ok(()); + } + + let query = build_topic_trace_root_metadata_query(project_id, &root_span_ids); + maybe_print_topic_query(print_queries, "trace-root-metadata", &query); + let response = execute_btql_value(client, &query).await?; + let metadata_by_root_span_id = btql_data_rows(&response) + .into_iter() + .filter_map(|row| { + let root_span_id = value_as_string(row.get("root_span_id"))?; + Some(( + root_span_id, + (trace_created_by_user_id(row), trace_git_origin_url(row)), + )) + }) + .collect::>(); + + for trace in traces { + if let Some((user_id, git_origin_url)) = metadata_by_root_span_id.get(&trace.root_span_id) { + if trace.created_by_user_id.is_none() { + trace.created_by_user_id = user_id.clone(); + } + if trace.git_origin_url.is_none() { + trace.git_origin_url = git_origin_url.clone(); + trace.repo = trace + .git_origin_url + .as_deref() + .and_then(repo_slug_from_origin_url); + } + } + } + + Ok(()) +} + +async fn fetch_org_users(client: &ApiClient) -> Result> { + let mut users = HashMap::new(); + let mut starting_after = None::; + + loop { + let mut path = format!( + "/v1/user?org_name={}&limit={ORG_USERS_PAGE_LIMIT}", + encode(client.org_name()) + ); + if let Some(cursor) = starting_after.as_deref() { + path.push_str("&starting_after="); + path.push_str(&encode(cursor)); + } + + let response: OrgUsersListResponse = client.get(&path).await?; + let objects = response.objects; + let page_len = objects.len(); + let next_cursor = objects.last().map(|user| user.id.clone()); + for user in objects { + users.insert(user.id.clone(), user); + } + + if page_len < ORG_USERS_PAGE_LIMIT { + break; + } + let Some(next_cursor) = next_cursor else { + break; + }; + if starting_after.as_deref() == Some(next_cursor.as_str()) { + break; + } + starting_after = Some(next_cursor); + } + + Ok(users) +} + +fn read_btql_f64_metric(row: &serde_json::Map, alias: &str) -> f64 { + value_as_f64(row.get(alias)).unwrap_or(0.0) +} + +fn metrics_total_tokens(metrics: Option<&Value>) -> Option { + let metrics = metrics?.as_object()?; + value_as_f64(metrics.get("total_tokens")) + .or_else(|| value_as_f64(metrics.get("tokens"))) + .or_else(|| { + let prompt = value_as_f64(metrics.get("prompt_tokens")) + .or_else(|| value_as_f64(metrics.get("input_tokens")))?; + let completion = value_as_f64(metrics.get("completion_tokens")) + .or_else(|| value_as_f64(metrics.get("output_tokens")))?; + Some(prompt + completion) + }) +} + +fn metrics_cost(metrics: Option<&Value>) -> Option { + let metrics = metrics?.as_object()?; + value_as_f64(metrics.get("estimated_cost")).or_else(|| value_as_f64(metrics.get("cost"))) +} + +fn metrics_duration_seconds(metrics: Option<&Value>) -> Option { + let metrics = metrics?.as_object()?; + value_as_f64(metrics.get("duration")).or_else(|| { + let start = value_as_f64(metrics.get("start"))?; + let end = value_as_f64(metrics.get("end"))?; + Some((end - start).max(0.0)) + }) +} + +fn value_as_f64(value: Option<&Value>) -> Option { + match value { + Some(Value::Number(number)) => number.as_f64(), + Some(Value::String(value)) => value.parse::().ok(), + _ => None, + } +} + +fn value_as_string(value: Option<&Value>) -> Option { + match value { + Some(Value::String(value)) => Some(value.clone()), + Some(Value::Number(value)) => Some(value.to_string()), + Some(Value::Bool(value)) => Some(value.to_string()), + _ => None, + } +} + +fn nested_value_as_string(row: &serde_json::Map, path: &[&str]) -> Option { + let (first, rest) = path.split_first()?; + let mut value = row.get(*first)?; + for key in rest { + value = value.as_object()?.get(*key)?; + } + value_as_string(Some(value)) +} + +fn format_preview_value(value: &Value) -> String { + match value { + Value::String(value) => value.clone(), + Value::Null => String::new(), + _ => serde_json::to_string(value).unwrap_or_else(|_| value.to_string()), + } +} + +fn combine_filter_clauses(clauses: I) -> String +where + I: IntoIterator>, +{ + let parts = clauses + .into_iter() + .flatten() + .map(|clause| clause.trim().to_string()) + .filter(|clause| !clause.is_empty()) + .map(|clause| format!("({clause})")) + .collect::>(); + if parts.is_empty() { + "true".to_string() + } else { + parts.join(" AND ") + } +} + +fn combine_optional_filter_clauses(clauses: I) -> Option +where + I: IntoIterator>, +{ + let combined = combine_filter_clauses(clauses); + if combined == "true" { + None + } else { + Some(combined) + } +} + +fn btql_string_literal(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + +fn btql_json_string_literal(value: &str) -> String { + serde_json::to_string(value) + .unwrap_or_else(|_| format!("\"{}\"", value.replace('\\', "\\\\").replace('\"', "\\\""))) +} + +fn maybe_print_topic_query(enabled: bool, label: &str, query: &str) { + if enabled { + eprintln!("bt topics [{label}] BTQL:\n{query}\n"); + } +} + +async fn fetch_topic_automation_progress( + client: &ApiClient, + project_id: &str, + time_filter_clause: &str, + cursor_status: &AutomationCursorSnapshot, + facet_bars: &[FacetStatusBar], + topic_bars: &[TopicStatusBar], + print_queries: bool, +) -> Result { + let pending_min_executed_xact_id = cursor_status.pending_min_executed_xact_id.as_deref(); + let mut measure_expressions = Vec::::new(); + + let facet_paths = facet_bars + .iter() + .map(|bar| bar.facet_path.as_str()) + .collect::>(); + if !facet_paths.is_empty() { + let facet_coverage_predicate = facet_paths + .iter() + .map(|facet_path| format!("({facet_path} != 'no_match')")) + .collect::>() + .join(" OR "); + measure_expressions.push(format!( + "count(({facet_coverage_predicate}) ? 1 : null) as facet_current_marked_traces" + )); + } + + for (index, bar) in facet_bars.iter().enumerate() { + let prefix = format!("facet_{index}"); + measure_expressions.push(format!( + "count((({} != 'no_match')) ? 1 : null) as {prefix}_current_marked_traces", + bar.facet_path + )); + measure_expressions.push(format!( + "count({}) as {prefix}_current_completed_output_traces", + bar.facet_path + )); + if bar.function_keys.is_empty() { + continue; + } + + let mut completed_predicates = Vec::new(); + let mut inflight_predicates = Vec::new(); + let mut error_predicates = Vec::new(); + for function_key in &bar.function_keys { + let predicates = + build_triggered_function_predicates(function_key, pending_min_executed_xact_id); + completed_predicates.push(format!("({})", predicates.completed_predicate)); + inflight_predicates.push(format!("({})", predicates.inflight_predicate)); + error_predicates.push(format!("({})", predicates.error_predicate)); + } + + measure_expressions.push(format!( + "count(({} ) ? 1 : null) as {prefix}_current_completed_traces", + completed_predicates.join(" OR ") + )); + measure_expressions.push(format!( + "count(({} ) ? 1 : null) as {prefix}_current_inflight_traces", + inflight_predicates.join(" OR ") + )); + measure_expressions.push(format!( + "count(({} ) ? 1 : null) as {prefix}_current_error_traces", + error_predicates.join(" OR ") + )); + } + + for (index, bar) in topic_bars.iter().enumerate() { + let prefix = format!("topic_{index}"); + measure_expressions.push(format!( + "count(({}) ? 1 : null) as {prefix}_current_eligible_traces", + bar.eligible_predicate + )); + measure_expressions.push(format!( + "count({}) as {prefix}_current_labeled_traces", + bar.classification_path + )); + if let Some(function_key) = bar.function_key.as_deref() { + let predicates = build_triggered_function_predicates(function_key, pending_min_executed_xact_id); measure_expressions.push(format!( "count(({} ) ? 1 : null) as {prefix}_current_completed_traces", @@ -1932,6 +3397,7 @@ async fn fetch_topic_automation_progress( let total_query = format!( "from: project_logs('{escaped_project_id}') spans | measures: count_distinct(root_span_id) as total_traces | filter: {time_filter_clause}" ); + maybe_print_topic_query(print_queries, "progress-total", &total_query); let total_response = execute_btql_value(client, &total_query).await?; let total_row = first_btql_row(&total_response); let aggregate_row = if measure_expressions.is_empty() { @@ -1941,6 +3407,7 @@ async fn fetch_topic_automation_progress( "from: project_logs('{escaped_project_id}') spans | measures: {} | filter: {time_filter_clause}", measure_expressions.join(", ") ); + maybe_print_topic_query(print_queries, "progress-counts", &aggregate_query); let aggregate_response = execute_btql_value(client, &aggregate_query).await?; first_btql_row(&aggregate_response).cloned() }; @@ -2546,4 +4013,283 @@ mod tests { ); assert_eq!(slugify_topic_map_name(" "), "topic-map"); } + + #[test] + fn topic_explore_time_filter_combines_window_and_extra_filter() { + let filter = combine_filter_clauses([ + Some(topic_explore_time_filter_clause(None, "6h").expect("time filter")), + Some("metadata.environment = 'test'".to_string()), + ]); + + assert_eq!( + filter, + "(created >= NOW() - INTERVAL 21600 SECOND) AND (metadata.environment = 'test')" + ); + } + + #[test] + fn topic_explore_filter_accepts_repo_shortcut_and_origin_urls() { + let origin = parse_repo_origin_selector("test-org/test-repo").expect("origin"); + let filter = topic_repo_origin_filter_clause(&origin, Some("test-org/test-repo")); + assert!(filter.contains("\"metadata\".\"git_origin_url\" IN [")); + assert!(filter.contains("'github.com/test-org/test-repo'")); + assert!(filter.contains("'test-org/test-repo'")); + assert!(filter.contains("'https://github.com/test-org/test-repo.git'")); + assert!(filter.contains("'https://github.com/test-org/test-repo'")); + assert!(filter.contains("'git@github.com:test-org/test-repo.git'")); + assert!(filter.contains("'ssh://git@github.com/test-org/test-repo.git'")); + + let query = build_topic_repo_root_spans_query( + "test-project", + "created >= NOW() - INTERVAL 604800 SECOND", + &filter, + 25, + Some("cursor-test"), + ); + assert!(query.contains("from: project_logs('test-project') spans")); + assert!(query.contains("created >= NOW() - INTERVAL 604800 SECOND")); + assert!(query.contains("span_id = root_span_id")); + assert!(query.contains("metadata.git_origin_url as git_origin_url")); + assert!(query.contains("cursor: \"cursor-test\"")); + + let https = + parse_repo_origin_selector("https://github.com/test-org/test-repo.git").expect("https"); + let ssh = parse_repo_origin_selector("git@github.com:test-org/test-repo.git") + .expect("ssh origin"); + let host_path = + parse_repo_origin_selector("github.com/test-org/test-repo").expect("host path"); + assert_eq!(https.canonical_slug(), "github.com/test-org/test-repo"); + assert_eq!(https, ssh); + assert_eq!(https, host_path); + } + + #[test] + fn topic_explore_filter_accepts_non_github_host() { + let origin = parse_repo_origin_selector("gitlab.example.com/platform/agent/runtime.git") + .expect("origin"); + + assert_eq!( + origin.canonical_slug(), + "gitlab.example.com/platform/agent/runtime" + ); + assert!(repo_origin_url_variants(&origin) + .contains("git@gitlab.example.com:platform/agent/runtime.git")); + } + + #[test] + fn topic_classifications_query_is_bounded_and_topic_source_scoped() { + let topic_map = TopicExploreTopicMap { + automation_id: "auto_test_topics".to_string(), + automation_name: "Topics".to_string(), + facet: Some("Task".to_string()), + topic_map: "Task".to_string(), + topic_map_id: "fn_test_topic_map".to_string(), + version: Some("123".to_string()), + classification_path: escape_btql_ident_path(&["classifications", "Task"]), + btql_filter: None, + }; + let filter = + topic_map_filter_clause(&topic_map, "created >= NOW() - INTERVAL 86400 SECOND"); + let query = build_topic_classifications_query( + "test-project", + "Task", + "fn_test_topic_map", + &filter, + TopicExploreSort::Cost, + 25, + ); + + assert!(query.contains("from: project_logs('test-project') summary")); + assert!(query.contains("created >= NOW() - INTERVAL 86400 SECOND")); + assert!(query.contains("\"classifications\".\"Task\" IS NOT NULL")); + assert!( + query.contains("\"classifications\".\"Task\".\"source\".\"id\" = 'fn_test_topic_map'") + ); + assert!(query.contains("sum(COALESCE(metrics.estimated_cost, metrics.cost, 0)) as cost")); + assert!(query.contains("sort: cost DESC")); + assert!(query.contains("limit: 25")); + } + + #[test] + fn topic_traces_query_filters_topic_id_and_sorts_tokens() { + let topic_map = TopicExploreTopicMap { + automation_id: "auto_test_topics".to_string(), + automation_name: "Topics".to_string(), + facet: Some("Task".to_string()), + topic_map: "Task".to_string(), + topic_map_id: "fn_test_topic_map".to_string(), + version: None, + classification_path: escape_btql_ident_path(&["classifications", "Task"]), + btql_filter: None, + }; + let mut filter = + topic_map_filter_clause(&topic_map, "created >= NOW() - INTERVAL 86400 SECOND"); + filter = combine_filter_clauses([ + Some(filter), + topic_filter_clause(&topic_map, None, Some("topic-test")).expect("topic filter"), + ]); + let query = build_topic_traces_query( + "test-project", + "Task", + "fn_test_topic_map", + &filter, + TopicExploreSort::Tokens, + 10, + Some("cursor-test"), + ); + + assert!(query.contains("select: created, root_span_id")); + assert!(query.contains("_pagination_key")); + assert!(query.contains("as sort_value")); + assert!(query.contains("span_attributes.created_by_user_id as created_by_user_id")); + assert!(query.contains("metadata.git_origin_url as git_origin_url")); + assert!(query.contains("\"classifications\".\"Task\".\"id\" = 'topic-test'")); + assert!(query.contains("sort: COALESCE(metrics.total_tokens, metrics.tokens")); + assert!(query.contains(", _pagination_key DESC")); + assert!(query.contains("limit: 10")); + assert!(query.contains("cursor: \"cursor-test\"")); + } + + #[test] + fn topic_trace_cursor_filters_after_last_visible_row() { + let traces = (0..11) + .map(|index| TopicTraceRow { + created: Some(format!("2026-07-27T12:{index:02}:00Z")), + root_span_id: format!("root-{index}"), + span_id: None, + row_id: None, + created_by_user_id: None, + created_by_user_name: None, + created_by_user_email: None, + git_origin_url: None, + repo: None, + topic: Some("Support".to_string()), + topic_id: Some("topic-test".to_string()), + tokens: (100 - index) as f64, + cost: 0.0, + duration_seconds: None, + input: None, + app_url: "https://example.com/app/org/p/project/logs".to_string(), + pagination_key: Some(format!("p{index:020}")), + sort_value: Some(TopicTraceCursorValue::Number((100 - index) as f64)), + }) + .collect::>(); + + let cursor = topic_trace_next_cursor(&traces, TopicExploreSort::Tokens, 10) + .expect("cursor") + .expect("full page has cursor"); + let decoded = parse_topic_trace_cursor(Some(&cursor), TopicExploreSort::Tokens) + .expect("parse cursor") + .expect("topic cursor"); + assert_eq!(decoded.pagination_key, "p00000000000000000009"); + assert_eq!(decoded.sort_value, TopicTraceCursorValue::Number(91.0)); + + let filter = + topic_trace_cursor_filter_clause(TopicExploreSort::Tokens, &decoded).expect("filter"); + assert!(filter.contains( + "COALESCE(metrics.total_tokens, metrics.tokens, metrics.prompt_tokens + metrics.completion_tokens, metrics.input_tokens + metrics.output_tokens, 0) < 91" + )); + assert!(filter.contains("_pagination_key < 'p00000000000000000009'")); + } + + #[test] + fn topic_trace_cursor_rejects_sort_mismatch() { + let cursor = encode_topic_trace_cursor(&TopicTracePaginationCursor { + version: 1, + sort: TopicExploreSort::Cost, + sort_value: TopicTraceCursorValue::Number(0.42), + pagination_key: "p00000000000000000009".to_string(), + }) + .expect("encode cursor"); + + let err = + parse_topic_trace_cursor(Some(&cursor), TopicExploreSort::Tokens).expect_err("err"); + assert!(err.to_string().contains("created with --sort cost")); + } + + #[test] + fn topic_trace_row_reads_created_by_user_id() { + let row = serde_json::json!({ + "created": "2026-07-27T12:00:00Z", + "root_span_id": "root-test", + "span_id": "span-test", + "created_by_user_id": "user-test", + "_pagination_key": "p00000000000000000010", + "sort_value": "15", + "metrics": { + "input_tokens": 10, + "output_tokens": 5, + "cost": 0.01 + }, + "git_origin_url": "git@github.com:test-org/test-repo.git" + }); + let row = row.as_object().expect("row object"); + + let trace = topic_trace_row_from_btql(row, "https://example.com/app/org/p/project/logs"); + + assert_eq!(trace.created_by_user_id.as_deref(), Some("user-test")); + assert_eq!(trace.created_by_user_name, None); + assert_eq!( + trace.git_origin_url.as_deref(), + Some("git@github.com:test-org/test-repo.git") + ); + assert_eq!(trace.repo.as_deref(), Some("github.com/test-org/test-repo")); + assert_eq!( + trace.pagination_key.as_deref(), + Some("p00000000000000000010") + ); + assert_eq!(trace.sort_value, Some(TopicTraceCursorValue::Number(15.0))); + assert_eq!(trace.tokens, 15.0); + } + + #[test] + fn topic_trace_row_reads_nested_created_by_user_id() { + let row = serde_json::json!({ + "root_span_id": "root-test", + "span_attributes": { + "created_by_user_id": "user-nested" + } + }); + let row = row.as_object().expect("row object"); + + let trace = topic_trace_row_from_btql(row, "https://example.com/app/org/p/project/logs"); + + assert_eq!(trace.created_by_user_id.as_deref(), Some("user-nested")); + } + + #[test] + fn topic_trace_root_metadata_query_is_root_span_bounded() { + let query = build_topic_trace_root_metadata_query( + "test-project", + &["root-one".to_string(), "root-two".to_string()], + ); + + assert!(query.contains("from: project_logs('test-project') spans")); + assert!(query.contains("root_span_id IN ['root-one', 'root-two']")); + assert!(query.contains("span_id = root_span_id")); + assert!(query.contains("span_attributes.created_by_user_id as created_by_user_id")); + assert!(query.contains("metadata.git_origin_url as git_origin_url")); + } + + #[test] + fn org_user_display_name_prefers_name_then_email() { + let named = OrgUser { + id: "user-named".to_string(), + given_name: Some("Ada".to_string()), + family_name: Some("Lovelace".to_string()), + email: Some("ada@example.com".to_string()), + }; + let emailed = OrgUser { + id: "user-emailed".to_string(), + given_name: Some(" ".to_string()), + family_name: None, + email: Some("trace-user@example.com".to_string()), + }; + + assert_eq!(named.display_name().as_deref(), Some("Ada Lovelace")); + assert_eq!( + emailed.display_name().as_deref(), + Some("trace-user@example.com") + ); + } } diff --git a/src/topics/explore.rs b/src/topics/explore.rs new file mode 100644 index 00000000..08e99648 --- /dev/null +++ b/src/topics/explore.rs @@ -0,0 +1,848 @@ +use std::fmt::Write as _; + +use anyhow::{bail, Result}; +use chrono::{DateTime, Utc}; + +use crate::{ + traces::{run_interactive_project_log_trace_list, ProjectLogTraceSeed}, + ui::{ + apply_column_padding, fuzzy_select_opt, header, is_interactive, print_with_pager, + styled_table, truncate, with_spinner, + }, +}; + +use super::{ + api::{self, TopicClassificationRow, TopicExploreFacet, TopicTraceRow}, + formatting::{format_count, format_project_header, format_timestamp_with_relative}, + ClassificationsArgs, ExploreArgs, ExploreTimeArgs, FacetsArgs, ResolvedContext, + TopicTracesArgs, +}; + +pub async fn run_facets(ctx: &ResolvedContext, args: &FacetsArgs, json: bool) -> Result<()> { + let filter_clause = resolve_filter_clause(ctx, &args.time, args.output.print_queries).await?; + let report = with_spinner( + "Loading facets and topic maps...", + api::fetch_topics_explore_facets( + ctx, + args.automation_id.as_deref(), + &filter_clause, + args.output.print_queries, + ), + ) + .await?; + + if json { + println!("{}", serde_json::to_string(&report)?); + return Ok(()); + } + + print_with_pager(&render_facets_report(&report))?; + Ok(()) +} + +pub async fn run_classifications( + ctx: &ResolvedContext, + args: &ClassificationsArgs, + json: bool, +) -> Result<()> { + let filter_clause = resolve_filter_clause(ctx, &args.time, args.output.print_queries).await?; + let report = with_spinner( + "Loading topic labels...", + api::fetch_topic_classifications( + ctx, + args.selection.automation_id.as_deref(), + args.selection.facet.as_deref(), + args.selection.topic_map.as_deref(), + args.sort_limit.sort, + args.sort_limit.limit, + &filter_clause, + args.output.print_queries, + ), + ) + .await?; + + if json { + println!("{}", serde_json::to_string(&report)?); + return Ok(()); + } + + print_with_pager(&render_classifications_report(&report))?; + Ok(()) +} + +pub async fn run_traces(ctx: &ResolvedContext, args: &TopicTracesArgs, json: bool) -> Result<()> { + let filter_clause = resolve_filter_clause(ctx, &args.time, args.output.print_queries).await?; + let report = with_spinner( + "Loading traces for topic label...", + api::fetch_topic_traces( + ctx, + args.selection.automation_id.as_deref(), + args.selection.facet.as_deref(), + args.selection.topic_map.as_deref(), + args.topic.topic.as_deref(), + args.topic.topic_id.as_deref(), + args.sort_limit.sort.into(), + args.sort_limit.limit, + args.cursor.as_deref(), + &filter_clause, + args.output.print_queries, + ), + ) + .await?; + + if json { + println!("{}", serde_json::to_string(&report)?); + return Ok(()); + } + + print_with_pager(&render_traces_report(&report))?; + Ok(()) +} + +pub async fn run_explore(ctx: &ResolvedContext, args: &ExploreArgs, json: bool) -> Result<()> { + if json { + bail!("`bt topics explore` is interactive and does not support --json; use `bt topics facets`, `bt topics classifications`, or `bt topics traces` with --json"); + } + if !is_interactive() { + bail!("`bt topics explore` requires a TTY; use `bt topics facets`, `bt topics classifications`, and `bt topics traces` for non-interactive exploration"); + } + if args.trace_page_size == 0 { + bail!("--trace-page-size must be greater than 0"); + } + + let filter_clause = resolve_filter_clause(ctx, &args.time, args.output.print_queries).await?; + let facets_report = with_spinner( + "Loading facets and topic maps...", + api::fetch_topics_explore_facets( + ctx, + args.selection.automation_id.as_deref(), + &filter_clause, + args.output.print_queries, + ), + ) + .await?; + if facets_report.facets.is_empty() { + bail!("no topic maps found; run `bt topics config` to inspect Topics setup"); + } + + let candidates = matching_explore_topic_maps(&facets_report.facets, args)?; + let mut default_facet_index = 0usize; + let mut force_facet_prompt = false; + let mut users_cache = api::OrgUsersCache::default(); + + while let Some((selected_map, selected_map_index)) = + select_explore_topic_map(&candidates, default_facet_index, force_facet_prompt)? + { + default_facet_index = selected_map_index; + let automation_id = Some(selected_map.automation_id.as_str()); + let facet = selected_map.facet.as_deref(); + let topic_map = Some(selected_map.topic_map_id.as_str()); + + let classifications_report = with_spinner( + "Loading topic labels...", + api::fetch_topic_classifications( + ctx, + automation_id, + facet, + topic_map, + args.sort_limit.sort, + args.sort_limit.limit, + &filter_clause, + args.output.print_queries, + ), + ) + .await?; + if classifications_report.classifications.is_empty() { + bail!("no topic labels found for the selected topic map in this time window"); + } + + let classification_labels = classifications_report + .classifications + .iter() + .map(format_classification_choice) + .collect::>(); + let mut default_classification_index = 0usize; + + loop { + let Some(classification_index) = fuzzy_select_opt( + "Select topic label: label / traces / cost / tokens / avg cost / id (Esc to facets)", + &classification_labels, + default_classification_index.min(classification_labels.len().saturating_sub(1)), + )? + else { + force_facet_prompt = true; + break; + }; + default_classification_index = classification_index; + let selected_classification = classifications_report + .classifications + .get(classification_index) + .expect("selected classification"); + + let topic_id = (!selected_classification.topic_id.is_empty()) + .then_some(selected_classification.topic_id.as_str()); + let topic = topic_id + .is_none() + .then_some(selected_classification.topic.as_str()); + + let mut traces_report = with_spinner( + "Loading matching traces...", + api::fetch_topic_traces_with_user_cache( + ctx, + automation_id, + facet, + topic_map, + topic, + topic_id, + args.sort_limit.sort, + args.trace_page_size, + None, + &filter_clause, + args.output.print_queries, + &mut users_cache, + ), + ) + .await?; + if traces_report.traces.is_empty() { + eprintln!( + "No traces found for the selected topic label in this time window. Select another topic label." + ); + continue; + } + + let topic_selection = ExploreTraceSelection { + automation_id, + facet, + topic_map, + topic, + topic_id, + filter_clause: &filter_clause, + }; + run_trace_picker( + ctx, + args, + topic_selection, + &mut traces_report, + &mut users_cache, + ) + .await?; + } + } + + Ok(()) +} + +async fn resolve_filter_clause( + ctx: &ResolvedContext, + time: &ExploreTimeArgs, + print_queries: bool, +) -> Result { + let filter = api::topic_explore_filter_clause( + ctx, + time.since.as_deref(), + &time.window, + time.filter.as_deref(), + time.repo.as_deref(), + print_queries, + ); + if time + .repo + .as_deref() + .map(str::trim) + .is_some_and(|repo| !repo.is_empty()) + { + with_spinner("Resolving repo filter...", filter).await + } else { + filter.await + } +} + +async fn run_trace_picker( + ctx: &ResolvedContext, + args: &ExploreArgs, + selection: ExploreTraceSelection<'_>, + traces_report: &mut api::TopicTracesReport, + users_cache: &mut api::OrgUsersCache, +) -> Result<()> { + let mut default_trace_index = 0usize; + + loop { + let trace_labels = trace_picker_labels(traces_report); + let Some(trace_index) = fuzzy_select_opt( + "Select trace: created / user / repo / cost / tokens / duration / root / input (Esc to topic labels)", + &trace_labels, + default_trace_index.min(trace_labels.len().saturating_sub(1)), + )? else { + break; + }; + + if trace_index == traces_report.traces.len() && traces_report.next_cursor.is_some() { + let previous_len = traces_report.traces.len(); + load_more_traces(ctx, args, selection, traces_report, users_cache).await?; + default_trace_index = previous_len.min(traces_report.traces.len().saturating_sub(1)); + continue; + } + + default_trace_index = trace_index; + let trace = traces_report + .traces + .get(trace_index) + .expect("selected trace"); + if trace.root_span_id.is_empty() { + println!("{}", render_selected_trace(trace)); + continue; + } + + let trace_seeds = traces_report + .traces + .iter() + .map(trace_viewer_seed) + .collect::>(); + run_interactive_project_log_trace_list( + ctx.client.clone(), + &ctx.project.id, + Some(&ctx.project.name), + trace_seeds.clone(), + &trace.root_span_id, + args.output.print_queries, + ) + .await?; + } + + Ok(()) +} + +#[derive(Clone, Copy)] +struct ExploreTraceSelection<'a> { + automation_id: Option<&'a str>, + facet: Option<&'a str>, + topic_map: Option<&'a str>, + topic: Option<&'a str>, + topic_id: Option<&'a str>, + filter_clause: &'a str, +} + +async fn load_more_traces( + ctx: &ResolvedContext, + args: &ExploreArgs, + selection: ExploreTraceSelection<'_>, + traces_report: &mut api::TopicTracesReport, + users_cache: &mut api::OrgUsersCache, +) -> Result<()> { + let Some(cursor) = traces_report.next_cursor.clone() else { + return Ok(()); + }; + + let next_report = with_spinner( + "Loading more traces...", + api::fetch_topic_traces_with_user_cache( + ctx, + selection.automation_id, + selection.facet, + selection.topic_map, + selection.topic, + selection.topic_id, + args.sort_limit.sort, + args.trace_page_size, + Some(&cursor), + selection.filter_clause, + args.output.print_queries, + users_cache, + ), + ) + .await?; + + traces_report.traces.extend(next_report.traces); + traces_report.next_cursor = next_report.next_cursor; + Ok(()) +} + +fn matching_explore_topic_maps( + rows: &[TopicExploreFacet], + args: &ExploreArgs, +) -> Result> { + let candidates = rows + .iter() + .filter(|row| { + optional_selector_matches( + Some(row.automation_id.as_str()), + args.selection.automation_id.as_deref(), + ) + }) + .filter(|row| { + optional_selector_matches( + Some(row.facet.as_deref().unwrap_or("Ungrouped")), + args.selection.facet.as_deref(), + ) + }) + .filter(|row| match args.selection.topic_map.as_deref() { + Some(selector) => { + selector_matches(&row.topic_map, selector) + || selector_matches(&row.topic_map_id, selector) + } + None => true, + }) + .cloned() + .collect::>(); + + if candidates.is_empty() { + bail!( + "topic map selection did not match any configured topic map; run `bt topics facets` to list available facets and topic maps" + ); + } + + Ok(candidates) +} + +fn select_explore_topic_map( + candidates: &[TopicExploreFacet], + default_index: usize, + force_prompt: bool, +) -> Result> { + match candidates.len() { + 0 => bail!("no topic maps to select from"), + 1 if !force_prompt => Ok(Some(( + candidates + .first() + .cloned() + .expect("single topic map candidate"), + 0, + ))), + _ => { + let labels = candidates + .iter() + .map(format_facet_choice) + .collect::>(); + let Some(index) = fuzzy_select_opt( + "Select facet/topic map: facet / topic map / labeled / eligible / errors / id (Esc to exit)", + &labels, + default_index.min(labels.len().saturating_sub(1)), + )? + else { + return Ok(None); + }; + Ok(Some(( + candidates + .get(index) + .cloned() + .expect("selected topic map candidate"), + index, + ))) + } + } +} + +fn optional_selector_matches(candidate: Option<&str>, selector: Option<&str>) -> bool { + match selector { + Some(selector) => candidate + .map(|candidate| selector_matches(candidate, selector)) + .unwrap_or(false), + None => true, + } +} + +fn selector_matches(candidate: &str, selector: &str) -> bool { + candidate == selector || candidate.eq_ignore_ascii_case(selector) +} + +fn render_facets_report(report: &api::TopicsExploreFacetsReport) -> String { + let mut output = format_project_header( + &report.project.name, + &report.project.id, + &report.project.org_name, + ); + output.push('\n'); + + if report.facets.is_empty() { + output.push_str("\nNo topic maps found. Run `bt topics config` to inspect Topics setup.\n"); + return output; + } + + writeln!( + output, + "{} facet/topic map rows found.", + format_count(report.facets.len()) + ) + .expect("write to string"); + + let mut table = styled_table(); + table.set_header(vec![ + header("Facet"), + header("Topic map"), + header("Topic map ID"), + header("Version"), + header("Eligible"), + header("Labeled"), + header("Processing"), + header("Errors"), + ]); + apply_column_padding(&mut table, (0, 6)); + + for row in &report.facets { + table.add_row(vec![ + row.facet.as_deref().unwrap_or("Ungrouped").to_string(), + row.topic_map.clone(), + row.topic_map_id.clone(), + row.version.as_deref().unwrap_or("-").to_string(), + format_count(row.eligible), + format_count(row.labeled), + format_count(row.processing), + format_count(row.errors), + ]); + } + + writeln!(output, "\n{table}").expect("write to string"); + output +} + +fn render_classifications_report(report: &api::TopicClassificationsReport) -> String { + let mut output = format_project_header( + &report.project.name, + &report.project.id, + &report.project.org_name, + ); + writeln!( + output, + "\nTopic map: {} / {} ({})", + report.topic_map.facet.as_deref().unwrap_or("Ungrouped"), + report.topic_map.topic_map, + report.topic_map.topic_map_id + ) + .expect("write to string"); + + if report.classifications.is_empty() { + output.push_str("\nNo topic labels found in this time window.\n"); + return output; + } + + writeln!( + output, + "{} topic labels found.", + format_count(report.classifications.len()) + ) + .expect("write to string"); + + let mut table = styled_table(); + table.set_header(vec![ + header("Topic label"), + header("Topic ID"), + header("Traces"), + header("Tokens"), + header("Cost"), + header("Avg tokens"), + header("Avg cost"), + header("Latest"), + ]); + apply_column_padding(&mut table, (0, 6)); + + for row in &report.classifications { + table.add_row(vec![ + truncate(&row.topic, 36), + truncate(&row.topic_id, 28), + format_count(row.traces), + format_tokens(row.tokens), + format_cost(row.cost), + format_tokens(row.avg_tokens), + format_cost(row.avg_cost), + row.latest + .as_deref() + .map(format_timestamp_with_relative) + .unwrap_or_else(|| "-".to_string()), + ]); + } + + writeln!(output, "\n{table}").expect("write to string"); + output +} + +fn render_traces_report(report: &api::TopicTracesReport) -> String { + let mut output = format_project_header( + &report.project.name, + &report.project.id, + &report.project.org_name, + ); + writeln!( + output, + "\nTopic map: {} / {} ({})", + report.topic_map.facet.as_deref().unwrap_or("Ungrouped"), + report.topic_map.topic_map, + report.topic_map.topic_map_id + ) + .expect("write to string"); + + if report.traces.is_empty() { + output.push_str("\nNo traces found in this time window.\n"); + return output; + } + + writeln!( + output, + "Showing {} traces.", + format_count(report.traces.len()) + ) + .expect("write to string"); + + let mut table = styled_table(); + table.set_header(vec![ + header("Created"), + header("Root span ID"), + header("User"), + header("Repo"), + header("Topic"), + header("Tokens"), + header("Cost"), + header("Duration"), + header("Input"), + ]); + apply_column_padding(&mut table, (0, 6)); + + for row in &report.traces { + table.add_row(vec![ + row.created + .as_deref() + .map(format_timestamp_with_relative) + .unwrap_or_else(|| "-".to_string()), + truncate(&row.root_span_id, 24), + trace_user_label(row) + .map(|user| truncate(&user, 28)) + .unwrap_or_else(|| "-".to_string()), + trace_repo_label(row) + .map(|repo| truncate(&repo, 28)) + .unwrap_or_else(|| "-".to_string()), + truncate(row.topic.as_deref().unwrap_or("-"), 28), + format_tokens(row.tokens), + format_cost(row.cost), + format_duration(row.duration_seconds), + row.input + .as_deref() + .map(|input| truncate(input, 70)) + .unwrap_or_else(|| "-".to_string()), + ]); + } + + writeln!(output, "\n{table}").expect("write to string"); + if let Some(cursor) = report.next_cursor.as_deref() { + writeln!( + output, + "\nNext cursor: {cursor}\nUse `bt topics traces --cursor ` with the same topic filters to fetch more traces." + ) + .expect("write to string"); + } + output +} + +fn trace_picker_labels(report: &api::TopicTracesReport) -> Vec { + let mut labels = report + .traces + .iter() + .map(format_trace_choice) + .collect::>(); + if report.next_cursor.is_some() { + labels.push(format!( + "{} {} {} {} {} {} {} {}", + left_cell("Load more traces", 16), + left_cell("", 24), + left_cell("", 24), + right_cell("", 9), + right_cell("", 10), + right_cell("", 8), + left_cell("", 24), + left_cell("", 60), + )); + } + labels +} + +fn format_facet_choice(row: &TopicExploreFacet) -> String { + format!( + "{} {} {} {} {} {}", + left_cell(row.facet.as_deref().unwrap_or("Ungrouped"), 18), + left_cell(&row.topic_map, 28), + right_cell(&format_count(row.labeled), 8), + right_cell(&format_count(row.eligible), 8), + right_cell(&format_count(row.errors), 6), + row.topic_map_id + ) +} + +fn format_classification_choice(row: &TopicClassificationRow) -> String { + format!( + "{} {} {} {} {} {}", + left_cell(&row.topic, 34), + right_cell(&format_count(row.traces), 8), + right_cell(&format_cost(row.cost), 9), + right_cell(&format_tokens(row.tokens), 10), + right_cell(&format_cost(row.avg_cost), 9), + row.topic_id + ) +} + +fn format_trace_choice(row: &TopicTraceRow) -> String { + format!( + "{} {} {} {} {} {} {} {}", + left_cell(&format_compact_timestamp(row.created.as_deref()), 16), + left_cell( + &trace_user_label(row).unwrap_or_else(|| "-".to_string()), + 24 + ), + left_cell( + &trace_repo_label(row).unwrap_or_else(|| "-".to_string()), + 24 + ), + right_cell(&format_cost(row.cost), 9), + right_cell(&format_tokens(row.tokens), 10), + right_cell(&format_duration(row.duration_seconds), 8), + left_cell(&row.root_span_id, 24), + row.input + .as_deref() + .map(|input| truncate(input, 60)) + .unwrap_or_else(|| "-".to_string()) + ) +} + +fn left_cell(value: &str, width: usize) -> String { + let value = truncate(value, width); + format!("{value: String { + let value = truncate(value, width); + format!("{value:>width$}") +} + +fn format_compact_timestamp(value: Option<&str>) -> String { + let Some(value) = value else { + return "-".to_string(); + }; + let Ok(parsed) = DateTime::parse_from_rfc3339(value) else { + return truncate(value, 16); + }; + parsed + .with_timezone(&Utc) + .format("%Y-%m-%d %H:%M") + .to_string() +} + +fn trace_viewer_seed(row: &TopicTraceRow) -> ProjectLogTraceSeed { + ProjectLogTraceSeed { + created: row.created.clone(), + root_span_id: row.root_span_id.clone(), + span_id: row.span_id.clone(), + row_id: row.row_id.clone(), + input: row.input.clone(), + duration_seconds: row.duration_seconds, + total_tokens: row.tokens, + estimated_cost: row.cost, + } +} + +fn render_selected_trace(trace: &TopicTraceRow) -> String { + let mut output = String::new(); + writeln!( + output, + "Selected trace: {}", + if trace.root_span_id.is_empty() { + "" + } else { + &trace.root_span_id + } + ) + .expect("write to string"); + if let Some(topic) = trace.topic.as_deref() { + writeln!(output, "topic: {topic}").expect("write to string"); + } + if let Some(created) = trace.created.as_deref() { + writeln!( + output, + "created: {}", + format_timestamp_with_relative(created) + ) + .expect("write to string"); + } + if let Some(user) = trace_user_label_with_id(trace) { + writeln!(output, "user: {user}").expect("write to string"); + } + if let Some(repo) = trace_repo_label(trace) { + writeln!(output, "repo: {repo}").expect("write to string"); + } + if let Some(origin) = trace.git_origin_url.as_deref() { + writeln!(output, "origin: {origin}").expect("write to string"); + } + writeln!(output, "url: {}", trace.app_url).expect("write to string"); + if !trace.root_span_id.is_empty() { + writeln!(output).expect("write to string"); + writeln!(output, "bt view trace --trace-id {}", trace.root_span_id) + .expect("write to string"); + writeln!(output, "bt view thread --trace-id {}", trace.root_span_id) + .expect("write to string"); + writeln!( + output, + "bt view waterfall --trace-id {}", + trace.root_span_id + ) + .expect("write to string"); + } + output +} + +fn trace_user_label(row: &TopicTraceRow) -> Option { + row.created_by_user_name + .as_deref() + .or(row.created_by_user_email.as_deref()) + .or(row.created_by_user_id.as_deref()) + .map(ToString::to_string) +} + +fn trace_user_label_with_id(row: &TopicTraceRow) -> Option { + let label = trace_user_label(row)?; + let Some(user_id) = row.created_by_user_id.as_deref() else { + return Some(label); + }; + if label == user_id { + Some(label) + } else { + Some(format!("{label} ({user_id})")) + } +} + +fn trace_repo_label(row: &TopicTraceRow) -> Option { + row.repo + .as_deref() + .or(row.git_origin_url.as_deref()) + .map(ToString::to_string) +} + +fn format_tokens(value: f64) -> String { + if !value.is_finite() || value <= 0.0 { + return "-".to_string(); + } + format_count(value.round() as usize) +} + +fn format_cost(value: f64) -> String { + if !value.is_finite() || value <= 0.0 { + return "-".to_string(); + } + if value < 0.001 { + return "<$0.001".to_string(); + } + if value < 1.0 { + return format!("${value:.3}"); + } + format!("${value:.2}") +} + +fn format_duration(seconds: Option) -> String { + let Some(seconds) = seconds.filter(|seconds| seconds.is_finite() && *seconds >= 0.0) else { + return "-".to_string(); + }; + if seconds < 1.0 { + return format!("{:.0}ms", seconds * 1000.0); + } + if seconds < 60.0 { + return format!("{seconds:.1}s"); + } + let minutes = (seconds / 60.0).floor(); + let remainder = seconds % 60.0; + format!("{minutes:.0}m {remainder:.0}s") +} diff --git a/src/topics/mod.rs b/src/topics/mod.rs index c5e67ded..5e411f9a 100644 --- a/src/topics/mod.rs +++ b/src/topics/mod.rs @@ -1,5 +1,5 @@ use anyhow::{bail, Result}; -use clap::{Args, Subcommand}; +use clap::{builder::BoolishValueParser, Args, Subcommand}; use std::path::PathBuf; use crate::{args::BaseArgs, project_context::resolve_project_command_context_with_auth_mode}; @@ -7,6 +7,7 @@ use crate::{args::BaseArgs, project_context::resolve_project_command_context_wit pub(crate) mod api; mod btmap; mod config; +mod explore; mod formatting; mod open; mod poke; @@ -23,6 +24,10 @@ Examples: bt topics status bt topics status --full bt topics status --watch + bt topics facets --window 7d + bt topics classifications --facet Task --repo test-org/test-repo --sort cost + bt topics traces --facet Task --topic-id --repo test-org/test-repo --sort tokens + bt topics explore --repo test-org/test-repo bt topics config bt topics config bt topics config enable @@ -49,6 +54,15 @@ enum TopicsCommands { Status(StatusArgs), /// View or edit Topics automation config Config(Box), + /// List active facets and topic maps for exploration + Facets(FacetsArgs), + /// List topic labels for a facet or topic map + #[command(visible_alias = "labels", alias = "classes")] + Classifications(ClassificationsArgs), + /// List traces matching a topic label + Traces(TopicTracesArgs), + /// Guided Topics exploration flow + Explore(ExploreArgs), /// Queue Topics to run on the next executor pass Poke, /// Rewind recent Topics history and queue it to reprocess @@ -80,6 +94,172 @@ struct StatusArgs { watch: bool, } +#[derive(Debug, Clone, Args)] +struct ExploreTimeArgs { + /// Relative time window, for example 1h or 7d + #[arg(long, env = "BT_TOPICS_WINDOW", default_value = "7d")] + window: String, + + /// Absolute lower bound timestamp (overrides --window) + #[arg(long, env = "BT_TOPICS_SINCE")] + since: Option, + + /// Additional BTQL filter expression + #[arg(long, env = "BT_TOPICS_FILTER")] + filter: Option, + + /// Git repository origin filter, for example owner/repo or github.com/owner/repo + #[arg(long, env = "BT_TOPICS_REPO", value_name = "REPO")] + repo: Option, +} + +#[derive(Debug, Clone, Args)] +struct ExploreSelectionArgs { + /// Specific automation ID to search within + #[arg(long = "automation-id", env = "BT_TOPICS_AUTOMATION_ID")] + automation_id: Option, + + /// Source facet name, for example Task + #[arg(long, env = "BT_TOPICS_FACET")] + facet: Option, + + /// Topic map name or function ID + #[arg(long = "topic-map", env = "BT_TOPICS_TOPIC_MAP")] + topic_map: Option, +} + +#[derive(Debug, Clone, Args)] +struct ExploreOutputArgs { + /// Print each BTQL query before execution + #[arg( + long = "print-queries", + env = "BT_TOPICS_PRINT_QUERIES", + value_parser = BoolishValueParser::new(), + default_value_t = false + )] + print_queries: bool, +} + +#[derive(Debug, Clone, Args)] +struct ExploreSortLimitArgs { + /// Number of rows to fetch + #[arg(long, env = "BT_TOPICS_LIMIT", default_value_t = 50)] + limit: usize, + + /// Sort topic labels by metric + #[arg( + long, + env = "BT_TOPICS_LABEL_SORT", + value_enum, + default_value = "count" + )] + sort: api::TopicExploreSort, +} + +#[derive(Debug, Clone, Args)] +struct TraceSortLimitArgs { + /// Number of rows to fetch + #[arg(long, env = "BT_TOPICS_LIMIT", default_value_t = 50)] + limit: usize, + + /// Sort trace rows by metric + #[arg( + long, + env = "BT_TOPICS_TRACE_SORT", + value_enum, + default_value = "recent" + )] + sort: api::TopicTraceSort, +} + +#[derive(Debug, Clone, Args)] +struct FacetsArgs { + /// Specific automation ID to search within + #[arg(long = "automation-id", env = "BT_TOPICS_AUTOMATION_ID")] + automation_id: Option, + + #[command(flatten)] + time: ExploreTimeArgs, + + #[command(flatten)] + output: ExploreOutputArgs, +} + +#[derive(Debug, Clone, Args)] +struct ClassificationsArgs { + #[command(flatten)] + selection: ExploreSelectionArgs, + + #[command(flatten)] + sort_limit: ExploreSortLimitArgs, + + #[command(flatten)] + time: ExploreTimeArgs, + + #[command(flatten)] + output: ExploreOutputArgs, +} + +#[derive(Debug, Clone, Args)] +struct TopicSelectorArgs { + /// Topic label to match + #[arg(long, env = "BT_TOPICS_TOPIC", conflicts_with = "topic_id")] + topic: Option, + + /// Stable topic ID to match + #[arg( + long = "topic-id", + env = "BT_TOPICS_TOPIC_ID", + conflicts_with = "topic" + )] + topic_id: Option, +} + +#[derive(Debug, Clone, Args)] +struct TopicTracesArgs { + #[command(flatten)] + selection: ExploreSelectionArgs, + + #[command(flatten)] + topic: TopicSelectorArgs, + + #[command(flatten)] + sort_limit: TraceSortLimitArgs, + + /// Cursor returned from a previous trace page + #[arg(long, env = "BT_TOPICS_CURSOR")] + cursor: Option, + + #[command(flatten)] + time: ExploreTimeArgs, + + #[command(flatten)] + output: ExploreOutputArgs, +} + +#[derive(Debug, Clone, Args)] +struct ExploreArgs { + #[command(flatten)] + selection: ExploreSelectionArgs, + + #[command(flatten)] + sort_limit: ExploreSortLimitArgs, + + /// Number of traces to fetch per page in the guided trace picker + #[arg( + long = "trace-page-size", + env = "BT_TOPICS_TRACE_PAGE_SIZE", + default_value_t = 10 + )] + trace_page_size: usize, + + #[command(flatten)] + time: ExploreTimeArgs, + + #[command(flatten)] + output: ExploreOutputArgs, +} + #[derive(Debug, Clone, Args)] struct ConfigArgs { /// Specific automation ID to show @@ -366,7 +546,13 @@ pub async fn run(base: BaseArgs, args: TopicsArgs) -> Result<()> { } let read_only = match args.command.as_ref() { - None | Some(TopicsCommands::Status(_)) | Some(TopicsCommands::Open) => true, + None + | Some(TopicsCommands::Status(_)) + | Some(TopicsCommands::Facets(_)) + | Some(TopicsCommands::Classifications(_)) + | Some(TopicsCommands::Traces(_)) + | Some(TopicsCommands::Explore(_)) + | Some(TopicsCommands::Open) => true, Some(TopicsCommands::Config(config_args)) => match config_args.command.as_ref() { None => true, Some(ConfigCommands::TopicMap(topic_map_args)) => { @@ -402,6 +588,18 @@ pub async fn run(base: BaseArgs, args: TopicsArgs) -> Result<()> { Some(TopicsCommands::Status(status_args)) => { status::run(&ctx, status_args, base.json).await } + Some(TopicsCommands::Facets(facets_args)) => { + explore::run_facets(&ctx, &facets_args, base.json).await + } + Some(TopicsCommands::Classifications(classifications_args)) => { + explore::run_classifications(&ctx, &classifications_args, base.json).await + } + Some(TopicsCommands::Traces(traces_args)) => { + explore::run_traces(&ctx, &traces_args, base.json).await + } + Some(TopicsCommands::Explore(explore_args)) => { + explore::run_explore(&ctx, &explore_args, base.json).await + } Some(TopicsCommands::Config(config_args)) => { let parent_automation_id = config_args.automation_id; let target = config_args.target; @@ -518,7 +716,13 @@ mod tests { fn topics_command_is_read_only(command: Option<&TopicsCommands>) -> bool { match command { - None | Some(TopicsCommands::Status(_)) | Some(TopicsCommands::Open) => true, + None + | Some(TopicsCommands::Status(_)) + | Some(TopicsCommands::Facets(_)) + | Some(TopicsCommands::Classifications(_)) + | Some(TopicsCommands::Traces(_)) + | Some(TopicsCommands::Explore(_)) + | Some(TopicsCommands::Open) => true, Some(TopicsCommands::Config(config_args)) => match config_args.command.as_ref() { None => true, Some(ConfigCommands::TopicMap(topic_map_args)) => { @@ -564,6 +768,109 @@ mod tests { let parsed = parse(&["topics", "report", "fn_123"]).expect("parse"); assert!(topics_command_is_read_only(parsed.command.as_ref())); + + let parsed = parse(&["topics", "facets"]).expect("parse"); + assert!(topics_command_is_read_only(parsed.command.as_ref())); + + let parsed = parse(&[ + "topics", + "classifications", + "--facet", + "Task", + "--sort", + "cost", + ]) + .expect("parse"); + assert!(topics_command_is_read_only(parsed.command.as_ref())); + + let parsed = parse(&[ + "topics", + "traces", + "--topic-map", + "fn_test_topic_map", + "--topic-id", + "topic-test", + ]) + .expect("parse"); + assert!(topics_command_is_read_only(parsed.command.as_ref())); + + let parsed = parse(&["topics", "explore"]).expect("parse"); + assert!(topics_command_is_read_only(parsed.command.as_ref())); + } + + #[test] + fn topics_explore_commands_parse_flags_and_aliases() { + let parsed = parse(&["topics", "facets"]).expect("parse"); + let Some(TopicsCommands::Facets(args)) = parsed.command.as_ref() else { + panic!("expected facets command"); + }; + assert_eq!(args.time.window, "7d"); + + let parsed = parse(&[ + "topics", + "labels", + "--automation-id", + "auto_test_topics", + "--facet", + "Task", + "--limit", + "25", + "--window", + "6h", + "--repo", + "test-org/test-repo", + "--filter", + "metadata.environment = 'test'", + "--print-queries", + ]) + .expect("parse"); + + let Some(TopicsCommands::Classifications(args)) = parsed.command.as_ref() else { + panic!("expected classifications command"); + }; + assert_eq!( + args.selection.automation_id.as_deref(), + Some("auto_test_topics") + ); + assert_eq!(args.selection.facet.as_deref(), Some("Task")); + assert_eq!(args.sort_limit.limit, 25); + assert_eq!(args.time.window, "6h"); + assert_eq!( + args.time.filter.as_deref(), + Some("metadata.environment = 'test'") + ); + assert_eq!(args.time.repo.as_deref(), Some("test-org/test-repo")); + assert!(args.output.print_queries); + + let parsed = parse(&[ + "topics", + "traces", + "--topic-map", + "fn_test_topic_map", + "--topic", + "Support", + "--sort", + "tokens", + "--cursor", + "cursor-test", + ]) + .expect("parse"); + let Some(TopicsCommands::Traces(args)) = parsed.command.as_ref() else { + panic!("expected traces command"); + }; + assert_eq!( + args.selection.topic_map.as_deref(), + Some("fn_test_topic_map") + ); + assert_eq!(args.topic.topic.as_deref(), Some("Support")); + assert_eq!(args.sort_limit.sort, api::TopicTraceSort::Tokens); + assert_eq!(args.cursor.as_deref(), Some("cursor-test")); + + let parsed = parse(&["topics", "explore", "--trace-page-size", "15"]).expect("parse"); + let Some(TopicsCommands::Explore(args)) = parsed.command.as_ref() else { + panic!("expected explore command"); + }; + assert_eq!(args.trace_page_size, 15); } #[test] diff --git a/src/traces.rs b/src/traces.rs index 6ef7ff1c..d32e9c76 100644 --- a/src/traces.rs +++ b/src/traces.rs @@ -301,6 +301,18 @@ struct ProjectSelection { name: Option, } +#[derive(Debug, Clone)] +pub(crate) struct ProjectLogTraceSeed { + pub(crate) created: Option, + pub(crate) root_span_id: String, + pub(crate) span_id: Option, + pub(crate) row_id: Option, + pub(crate) input: Option, + pub(crate) duration_seconds: Option, + pub(crate) total_tokens: f64, + pub(crate) estimated_cost: f64, +} + #[derive(Debug, Clone)] struct ParsedTraceUrl { org: Option, @@ -1637,6 +1649,125 @@ async fn run_interactive_trace_target( .await } +pub(crate) async fn run_interactive_project_log_trace_list( + client: ApiClient, + project_id: &str, + project_name: Option<&str>, + traces: Vec, + selected_root_span_id: &str, + print_queries: bool, +) -> Result<()> { + let selected_root_span_id = selected_root_span_id.trim(); + if selected_root_span_id.is_empty() { + bail!("selected trace is missing root_span_id"); + } + + let object_ref = ObjectRef { + object_type: "project_logs".to_string(), + object_name: project_id.to_string(), + }; + let source_expr = btql_source_expr(&object_ref)?; + let base_filter = root_span_filter_for_trace_seeds(&traces, selected_root_span_id); + let startup_trace_url = ParsedTraceUrl { + org: Some(client.org_name().to_string()), + project: Some(project_id.to_string()), + page: Some("logs".to_string()), + experiment: None, + comparison_experiment: None, + row_ref: Some(selected_root_span_id.to_string()), + span_id: Some(selected_root_span_id.to_string()), + trace_view_type: detail_view_trace_url_value(DetailView::Span).map(ToString::to_string), + }; + + run_interactive( + InteractiveRunArgs { + init: TraceViewerInit { + project: ProjectSelection { + id: project_id.to_string(), + name: project_name.map(ToString::to_string), + }, + source_expr, + list_mode: ListMode::Summary, + base_filter, + traces: trace_seed_summary_rows(traces), + limit: 100, + preview_length: 125, + print_queries, + }, + initial_search_query: String::new(), + startup_trace_url: Some(startup_trace_url), + }, + client, + ) + .await +} + +fn root_span_filter_for_trace_seeds( + traces: &[ProjectLogTraceSeed], + selected_root_span_id: &str, +) -> String { + let mut root_span_ids = BTreeSet::new(); + root_span_ids.insert(selected_root_span_id.to_string()); + for trace in traces { + let root_span_id = trace.root_span_id.trim(); + if !root_span_id.is_empty() { + root_span_ids.insert(root_span_id.to_string()); + } + } + + let clauses = root_span_ids + .into_iter() + .map(|root_span_id| format!("root_span_id = {}", sql_quote(&root_span_id))) + .collect::>(); + match clauses.as_slice() { + [] => "root_span_id = ''".to_string(), + [single] => single.clone(), + _ => format!("({})", clauses.join(" OR ")), + } +} + +fn trace_seed_summary_rows(traces: Vec) -> Vec { + traces.into_iter().map(trace_seed_summary_row).collect() +} + +fn trace_seed_summary_row(seed: ProjectLogTraceSeed) -> TraceSummaryRow { + let mut row = Map::new(); + insert_string_if_present(&mut row, "created", seed.created); + insert_string_if_present(&mut row, "root_span_id", Some(seed.root_span_id.clone())); + insert_string_if_present(&mut row, "span_id", seed.span_id); + insert_string_if_present(&mut row, "id", seed.row_id); + insert_string_if_present(&mut row, "input", seed.input); + + let mut metrics = Map::new(); + if let Some(duration) = seed.duration_seconds { + insert_number_if_finite(&mut metrics, "duration", duration); + } + insert_number_if_finite(&mut metrics, "total_tokens", seed.total_tokens); + insert_number_if_finite(&mut metrics, "estimated_cost", seed.estimated_cost); + if !metrics.is_empty() { + row.insert("metrics".to_string(), Value::Object(metrics)); + } + + TraceSummaryRow { + root_span_id: seed.root_span_id, + row, + } +} + +fn insert_string_if_present(row: &mut Map, key: &str, value: Option) { + let Some(value) = value.filter(|value| !value.is_empty()) else { + return; + }; + row.insert(key.to_string(), Value::String(value)); +} + +fn insert_number_if_finite(row: &mut Map, key: &str, value: f64) { + let Some(number) = serde_json::Number::from_f64(value).filter(|_| value.is_finite()) else { + return; + }; + row.insert(key.to_string(), Value::Number(number)); +} + async fn resolve_object_ref_for_view( client: &ApiClient, base: &BaseArgs, @@ -6755,6 +6886,36 @@ mod tests { } } + #[test] + fn project_log_trace_seed_rows_keep_topic_trace_context() { + let seeds = vec![ProjectLogTraceSeed { + created: Some("2026-07-27T12:00:00Z".to_string()), + root_span_id: "root-topic".to_string(), + span_id: Some("span-topic".to_string()), + row_id: Some("row-topic".to_string()), + input: Some("topic trace input".to_string()), + duration_seconds: Some(1.25), + total_tokens: 42.0, + estimated_cost: 0.002, + }]; + + let filter = root_span_filter_for_trace_seeds(&seeds, "root-selected"); + assert!(filter.contains("root_span_id = 'root-selected'")); + assert!(filter.contains("root_span_id = 'root-topic'")); + + let rows = trace_seed_summary_rows(seeds); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].root_span_id, "root-topic"); + assert_eq!( + value_as_string(rows[0].row.get("input")).as_deref(), + Some("topic trace input") + ); + assert_eq!( + extract_duration_seconds(rows[0].row.get("metrics")), + Some(1.25) + ); + } + fn draw_messages_once(messages: &[Value], expanded: bool) -> bool { let backend = TestBackend::new(140, 40); let mut terminal = Terminal::new(backend).expect("create terminal"); diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 3f51e64d..caf35bad 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -56,7 +56,7 @@ pub use ratatui_table::{ box_with_title, render_experiment_summary_table, summary_metric_unit, SummaryExperimentColumn, SummaryMetricCell, SummaryMetricKind, SummaryMetricRow, SummaryTableOptions, }; -pub use select::{fuzzy_select, select_project, ProjectSelectMode}; +pub use select::{fuzzy_select, fuzzy_select_opt, select_project, ProjectSelectMode}; pub use spinner::{with_spinner, with_spinner_visible}; pub use status::{print_command_status, CommandStatus}; diff --git a/topics-explore.md b/topics-explore.md new file mode 100644 index 00000000..8dc15f67 --- /dev/null +++ b/topics-explore.md @@ -0,0 +1,123 @@ +# Topics Explore Spec + +## Goal + +Add read-only Topics exploration commands that bridge automation status and trace inspection: + +```bash +bt topics facets +bt topics classifications +bt topics traces +bt topics explore +``` + +## Commands + +```bash +bt topics facets --window 7d +``` + +Lists active facets and topic maps for the current project. + +Columns: `facet`, `topic_map`, `topic_map_id`, `version`, `eligible`, `labeled`, `processing`, `errors`. + +```bash +bt topics classifications --facet Task --repo test-org/test-repo --sort cost --window 7d +``` + +Lists current topic labels for one facet or topic map. The command name follows the underlying `classifications` BTQL field. When `--repo` is provided, label counts and metrics are scoped to that repository. + +Text columns: `Topic label`, `Topic ID`, `Traces`, `Tokens`, `Cost`, `Avg tokens`, `Avg cost`, `Latest`. + +```bash +bt topics traces --facet Task --topic-id --repo test-org/test-repo --sort cost --window 7d +``` + +Lists traces matching a selected topic label. + +Columns: `created`, `root_span_id`, `user`, `repo`, `topic`, `tokens`, `cost`, `duration`, `input`. + +```bash +bt topics explore +``` + +Guided flow: + +1. Select automation if multiple exist. +2. Select facet/topic map. +3. Browse topic labels, sortable by count/tokens/cost. +4. Select a topic label to browse matching traces. +5. Select a trace to open the interactive trace view. +6. Quit the trace view to return to the topic-filtered trace list and keep exploring. +7. Select "Load more traces" to fetch the next trace page when available. +8. Press Esc on the trace list to move back up to the topic-label list. +9. Press Esc on the topic-label list to move back up to the facet/topic-map selector. + +## Shared Flags + +```bash +--automation-id +--facet +--topic-map +--topic