From cc56e12175edcc7837328f6bad42dc09c6588d8e Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Tue, 28 Jul 2026 10:57:39 -0700 Subject: [PATCH 1/2] topics explore --- src/topics/api.rs | 1419 ++++++++++++++++++++++++++++++++++++++++- src/topics/explore.rs | 815 +++++++++++++++++++++++ src/topics/mod.rs | 296 ++++++++- src/traces.rs | 161 +++++ src/ui/mod.rs | 2 +- topics-explore.md | 105 +++ 6 files changed, 2792 insertions(+), 6 deletions(-) create mode 100644 src/topics/explore.rs create mode 100644 topics-explore.md diff --git a/src/topics/api.rs b/src/topics/api.rs index 736276ee..aee8e3df 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,8 @@ 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_TRACE_CURSOR_PREFIX: &str = "bt-topic-traces-v1:"; #[derive(Debug, Clone, Serialize)] pub struct TopicsStatusReport { @@ -56,6 +59,243 @@ 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 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)] +#[serde(untagged)] +enum OrgUsersListResponse { + Envelope { objects: Vec }, + Bare(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")) + } +} + +impl OrgUsersListResponse { + fn into_objects(self) -> Vec { + match self { + Self::Envelope { objects } => objects, + Self::Bare(objects) => objects, + } + } +} + #[derive(Debug, Clone, Serialize)] struct TopicMapReportUrlRequest<'a> { function_id: &'a str, @@ -399,6 +639,272 @@ pub async fn fetch_topics_status( }) } +pub fn topic_explore_time_filter_clause( + since: Option<&str>, + window: &str, + extra_filter: Option<&str>, +) -> Result { + let time_clause = 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") + }; + + Ok(combine_filter_clauses([ + Some(time_clause), + extra_filter + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string), + ])) +} + +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); + } + hydrate_trace_root_created_by_user_ids( + &ctx.client, + &ctx.project.id, + &mut traces, + print_queries, + ) + .await + .context("failed to resolve trace root users")?; + users_cache + .hydrate_trace_users(&ctx.client, &mut traces) + .await + .context("failed to resolve trace users")?; + + 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 +1779,7 @@ async fn build_topic_automation_status( &cursor, &facet_bars, &topic_bars, + false, ) .await?; total_traces = progress.total_traces; @@ -1835,6 +2342,692 @@ fn ensure_facet_bar( } } +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 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::>(); + + 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" + ) + } + } +} + +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(); + + 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, + }); + } + } + + 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) + ))) + } + } +} + +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, {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_users_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 | 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)); + } + + 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, + 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()) +} + +async fn hydrate_trace_root_created_by_user_ids( + 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()) + .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_users_query(project_id, &root_span_ids); + maybe_print_topic_query(print_queries, "trace-users", &query); + let response = execute_btql_value(client, &query).await?; + let users_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"))?; + let user_id = trace_created_by_user_id(row)?; + Some((root_span_id, user_id)) + }) + .collect::>(); + + for trace in traces { + if trace.created_by_user_id.is_some() { + continue; + } + if let Some(user_id) = users_by_root_span_id.get(&trace.root_span_id) { + trace.created_by_user_id = Some(user_id.clone()); + } + } + + 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.into_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, @@ -1842,6 +3035,7 @@ async fn fetch_topic_automation_progress( 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(); @@ -1932,6 +3126,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 +3136,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 +3742,223 @@ mod tests { ); assert_eq!(slugify_topic_map_name(" "), "topic-map"); } + + #[test] + fn topic_explore_time_filter_combines_window_and_extra_filter() { + let filter = + topic_explore_time_filter_clause(None, "6h", Some("metadata.environment = 'test'")) + .expect("filter"); + + assert_eq!( + filter, + "(created >= NOW() - INTERVAL 21600 SECOND) AND (metadata.environment = 'test')" + ); + } + + #[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("\"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, + 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 + } + }); + 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.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_users_query_is_root_span_bounded() { + let query = build_topic_trace_root_users_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")); + } + + #[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..fd0039e8 --- /dev/null +++ b/src/topics/explore.rs @@ -0,0 +1,815 @@ +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, FacetsArgs, ResolvedContext, TopicTracesArgs, +}; + +pub async fn run_facets(ctx: &ResolvedContext, args: &FacetsArgs, json: bool) -> Result<()> { + let filter_clause = api::topic_explore_time_filter_clause( + args.time.since.as_deref(), + &args.time.window, + args.time.filter.as_deref(), + )?; + 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 = api::topic_explore_time_filter_clause( + args.time.since.as_deref(), + &args.time.window, + args.time.filter.as_deref(), + )?; + 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 = api::topic_explore_time_filter_clause( + args.time.since.as_deref(), + &args.time.window, + args.time.filter.as_deref(), + )?; + 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 = api::topic_explore_time_filter_clause( + args.time.since.as_deref(), + &args.time.window, + args.time.filter.as_deref(), + )?; + 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 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 / 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, 3)); + + 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, 3)); + + 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("Topic"), + header("Tokens"), + header("Cost"), + header("Duration"), + header("Input"), + ]); + apply_column_padding(&mut table, (0, 3)); + + 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()), + 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("", 26), + right_cell("", 9), + right_cell("", 10), + right_cell("", 8), + left_cell("", 24), + )); + } + 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()), + 26 + ), + 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"); + } + 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 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..c9c54ccc 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 --sort cost + bt topics traces --facet Task --topic-id --sort tokens + bt topics explore 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,158 @@ 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, +} + +#[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_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_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 +532,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 +574,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 +702,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 +754,106 @@ 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", + "--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!(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..c59b230c --- /dev/null +++ b/topics-explore.md @@ -0,0 +1,105 @@ +# 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 --sort cost --window 7d +``` + +Lists current topic labels for one facet or topic map. The command name follows the underlying `classifications` BTQL field. + +Text columns: `Topic label`, `Topic ID`, `Traces`, `Tokens`, `Cost`, `Avg tokens`, `Avg cost`, `Latest`. + +```bash +bt topics traces --facet Task --topic-id --sort cost --window 7d +``` + +Lists traces matching a selected topic label. + +Columns: `created`, `root_span_id`, `user`, `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