From 35f6c556ff7d919442c35e4f636abfa434b2b202 Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 26 Aug 2026 15:36:46 -0700 Subject: [PATCH 1/4] feat: add active observability templates --- src/active_observability_template/mod.rs | 333 +++++++ src/active_observability_template/pull.rs | 209 +++++ src/active_observability_template/push.rs | 732 +++++++++++++++ src/active_observability_template/template.rs | 832 ++++++++++++++++++ src/functions/api.rs | 45 + src/functions/mod.rs | 3 + src/http.rs | 20 + src/main.rs | 11 + src/topics/api.rs | 57 ++ 9 files changed, 2242 insertions(+) create mode 100644 src/active_observability_template/mod.rs create mode 100644 src/active_observability_template/pull.rs create mode 100644 src/active_observability_template/push.rs create mode 100644 src/active_observability_template/template.rs diff --git a/src/active_observability_template/mod.rs b/src/active_observability_template/mod.rs new file mode 100644 index 00000000..a9e6be38 --- /dev/null +++ b/src/active_observability_template/mod.rs @@ -0,0 +1,333 @@ +mod pull; +mod push; +mod template; + +use std::path::PathBuf; + +use anyhow::{bail, Context, Result}; +use clap::{Args, Subcommand}; +use dialoguer::{theme::ColorfulTheme, Confirm}; + +use crate::{ + args::BaseArgs, + functions::api::list_all_functions, + http::{build_http_client, DEFAULT_HTTP_TIMEOUT}, + project_context::resolve_project_command_context_with_auth_mode, + topics::api::list_project_automations, + ui::{self, print_command_status, with_spinner, CommandStatus}, + utils::read_text_source, +}; + +use self::{push::Snapshot, template::ActiveObservabilityTemplate}; + +#[derive(Debug, Clone, Args)] +#[command(after_help = "\ +Examples: + bt active-observability-template pull --output active-observability-template.json + bt active-observability-template push active-observability-template.json --project test-project + bt active-observability-template push https://example.com/active-observability-template.json + bt active-observability-template pull | bt active-observability-template push - --project test-project +")] +pub(crate) struct ActiveObservabilityTemplateArgs { + #[command(subcommand)] + command: ActiveObservabilityTemplateCommand, +} + +#[derive(Debug, Clone, Subcommand)] +enum ActiveObservabilityTemplateCommand { + /// Pull facets and Loop automations into a portable template + Pull(PullArgs), + /// Push facets and Loop automations from a portable template + Push(PushArgs), +} + +#[derive(Debug, Clone, Args)] +pub(super) struct PullArgs { + /// Write the template to this path instead of stdout + #[arg( + long, + short = 'O', + env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PULL_OUTPUT", + value_name = "PATH" + )] + output: Option, + + /// Overwrite an existing output file + #[arg( + long, + env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PULL_FORCE", + default_value_t = false, + value_parser = clap::builder::BoolishValueParser::new() + )] + force: bool, +} + +#[derive(Debug, Clone, Args)] +struct PushArgs { + /// Template path, HTTP(S) URL, or - to read from stdin + #[arg(value_name = "SOURCE")] + source_positional: Option, + + /// Template path, HTTP(S) URL, or - to read from stdin + #[arg( + long = "file", + short = 'f', + env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_FILE", + value_name = "SOURCE" + )] + source_flag: Option, + + /// Use this existing Topics automation for every facet + #[arg( + long, + env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_TOPICS_AUTOMATION", + value_name = "NAME_OR_ID" + )] + topics_automation: Option, + + /// Replace existing matching resources + #[arg( + long, + env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_FORCE", + default_value_t = false, + value_parser = clap::builder::BoolishValueParser::new() + )] + force: bool, + + /// Skip the confirmation prompt + #[arg( + long, + short = 'y', + env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_YES", + default_value_t = false, + value_parser = clap::builder::BoolishValueParser::new() + )] + yes: bool, +} + +impl PushArgs { + fn source(&self) -> Result<&str> { + match (&self.source_positional, &self.source_flag) { + (Some(_), Some(_)) => bail!("use either a template source or --file, not both"), + (Some(source), None) | (None, Some(source)) => Ok(source), + (None, None) => bail!( + "active observability template source required. Use: bt active-observability-template push " + ), + } + } +} + +pub(crate) async fn run(base: BaseArgs, args: ActiveObservabilityTemplateArgs) -> Result<()> { + match args.command { + ActiveObservabilityTemplateCommand::Pull(args) => pull::run(base, args).await, + ActiveObservabilityTemplateCommand::Push(args) => run_push(base, args).await, + } +} + +async fn run_push(base: BaseArgs, args: PushArgs) -> Result<()> { + let template = with_spinner( + "Loading active observability template...", + read_template_source(args.source()?), + ) + .await?; + template::validate(&template)?; + + let ctx = resolve_project_command_context_with_auth_mode(&base, false).await?; + let snapshot = with_spinner("Checking target resources...", async { + let (functions, automations) = tokio::try_join!( + list_all_functions(&ctx.client, &ctx.project.id), + list_project_automations(&ctx.client, &ctx.project.id), + )?; + Ok::<_, anyhow::Error>(Snapshot { + functions, + automations, + }) + }) + .await?; + let plan = push::plan( + &template, + snapshot, + args.topics_automation.as_deref(), + args.force, + )?; + + if !args.yes && ui::is_interactive() && !confirm_push(&ctx, &template, args.force)? { + return Ok(()); + } + + let result = with_spinner( + "Pushing active observability template...", + push::execute(&ctx.client, &ctx.project.id, plan), + ) + .await?; + + if base.json { + println!( + "{}", + serde_json::to_string(&serde_json::json!({ + "kind": "active_observability_template", + "status": "pushed", + "project": ctx.project.name, + "facets": result.facets, + "automations": result.automations, + }))? + ); + } else { + print_command_status( + CommandStatus::Success, + &format!( + "Pushed active observability template to '{}' ({} facets, {} Loop automations)", + ctx.project.name, + result.facets.len(), + result.automations.len() + ), + ); + } + Ok(()) +} + +fn confirm_push( + ctx: &crate::project_context::ProjectContext, + template: &ActiveObservabilityTemplate, + force: bool, +) -> Result { + let replacement = if force { + " and replace matching resources" + } else { + "" + }; + let prompt = format!( + "Push {} facets and {} Loop automations, including Topics wiring, to {}/{}{}?", + template.facets.len(), + template.automations.len(), + ctx.client.org_name(), + ctx.project.name, + replacement + ); + let term = + ui::prompt_term().ok_or_else(|| anyhow::anyhow!("interactive mode requires a TTY"))?; + Confirm::with_theme(&ColorfulTheme::default()) + .with_prompt(prompt) + .default(false) + .interact_on(&term) + .context("failed to confirm active observability template push") +} + +async fn read_template_source(source: &str) -> Result { + let contents = if source == "-" { + read_text_source(source, "active observability template")? + } else if source.starts_with("http://") || source.starts_with("https://") { + let url = reqwest::Url::parse(source).context("invalid template URL")?; + let response = build_http_client(DEFAULT_HTTP_TIMEOUT)? + .get(url) + .send() + .await + .context("failed to download template URL")? + .error_for_status() + .context("failed to download template URL")?; + response + .text() + .await + .context("failed to read template URL response")? + } else { + std::fs::read_to_string(source) + .with_context(|| format!("failed to read active observability template {source}"))? + }; + + serde_json::from_str(&contents).with_context(|| { + if source == "-" { + "failed to parse active observability template from stdin as JSON".to_string() + } else if source.starts_with("http://") || source.starts_with("https://") { + "failed to parse template URL as JSON; for GitHub Gists, use the Raw URL".to_string() + } else { + format!("failed to parse active observability template {source} as JSON") + } + }) +} + +#[cfg(test)] +mod tests { + use clap::Parser; + + use super::*; + + #[test] + fn active_observability_commands_parse_from_the_root_cli() { + for args in [ + vec!["bt", "active-observability-template", "pull"], + vec![ + "bt", + "active-observability-template", + "pull", + "--output", + "template.json", + ], + vec![ + "bt", + "active-observability-template", + "push", + "template.json", + "--topics-automation", + "Topics", + "--force", + "--yes", + ], + vec![ + "bt", + "active-observability-template", + "push", + "https://example.com/template.json", + ], + vec![ + "bt", + "active-observability-template", + "push", + "--file", + "template.json", + ], + ] { + crate::Cli::try_parse_from(args).expect("command should parse"); + } + } + + #[test] + fn active_observability_push_requires_exactly_one_source() { + let neither = PushArgs { + source_positional: None, + source_flag: None, + topics_automation: None, + force: false, + yes: false, + }; + assert!(neither + .source() + .unwrap_err() + .to_string() + .contains("source required")); + + let both = PushArgs { + source_positional: Some("one.json".to_string()), + source_flag: Some("two.json".to_string()), + topics_automation: None, + force: false, + yes: false, + }; + assert!(both.source().unwrap_err().to_string().contains("either")); + } + + #[tokio::test] + async fn active_observability_reads_a_local_template() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("template.json"); + std::fs::write( + &path, + r#"{"kind":"active_observability_template","schema_version":1}"#, + ) + .expect("write template"); + + let template = read_template_source(path.to_str().expect("UTF-8 path")) + .await + .expect("read template"); + assert_eq!(template.kind, template::KIND); + } +} diff --git a/src/active_observability_template/pull.rs b/src/active_observability_template/pull.rs new file mode 100644 index 00000000..b6cc0173 --- /dev/null +++ b/src/active_observability_template/pull.rs @@ -0,0 +1,209 @@ +use std::collections::HashSet; +use std::path::Path; + +use anyhow::{bail, Context, Result}; +use dialoguer::{theme::ColorfulTheme, MultiSelect}; + +use crate::{ + args::BaseArgs, + functions::api::list_all_functions, + project_context::resolve_project_command_context_with_auth_mode, + topics::api::list_project_automations, + ui::{self, print_command_status, with_spinner, CommandStatus}, + utils::write_json_atomic, +}; + +use super::{ + template::{ + deduplicate_preprocessors, from_remote, ActiveObservabilityTemplate, AutomationTemplate, + FacetTemplate, + }, + PullArgs, +}; + +pub(crate) async fn run(base: BaseArgs, args: PullArgs) -> Result<()> { + let ctx = resolve_project_command_context_with_auth_mode(&base, true).await?; + let (functions, automations) = + with_spinner("Loading active observability resources...", async { + tokio::try_join!( + list_all_functions(&ctx.client, &ctx.project.id), + list_project_automations(&ctx.client, &ctx.project.id), + ) + }) + .await?; + let mut template = from_remote(&functions, &automations)?; + if !base.json && !base.no_input && ui::is_interactive() { + (template.facets, template.automations) = + select_resources(template.facets, template.automations)?; + } + // Selection happens first so a selected facet never loses its required definition. + deduplicate_preprocessors(&mut template.facets); + + write_template(&template, args.output.as_deref(), args.force)?; + if let Some(path) = args + .output + .as_deref() + .filter(|path| *path != Path::new("-")) + { + if base.json { + println!( + "{}", + serde_json::to_string(&serde_json::json!({ + "kind": "active_observability_template", + "status": "pulled", + "project": ctx.project.name, + "output": path, + "facet_count": template.facets.len(), + "automation_count": template.automations.len(), + }))? + ); + } else { + print_command_status( + CommandStatus::Success, + &format!( + "Pulled active observability template from '{}' to {} ({} facets, {} Loop automations)", + ctx.project.name, + path.display(), + template.facets.len(), + template.automations.len() + ), + ); + } + } + Ok(()) +} + +fn write_template( + template: &ActiveObservabilityTemplate, + output: Option<&Path>, + force: bool, +) -> Result<()> { + match output { + Some(path) if path != Path::new("-") => { + if !force + && path + .try_exists() + .with_context(|| format!("failed to check {}", path.display()))? + { + bail!( + "output file {} already exists; use --force to overwrite it", + path.display() + ); + } + write_json_atomic(path, template) + } + _ => { + println!("{}", serialize_stdout(template)?); + Ok(()) + } + } +} + +fn serialize_stdout(template: &ActiveObservabilityTemplate) -> Result { + serde_json::to_string_pretty(template).context("failed to serialize template") +} + +fn select_resources( + facets: Vec, + automations: Vec, +) -> Result<(Vec, Vec)> { + if facets.is_empty() && automations.is_empty() { + return Ok((facets, automations)); + } + let labels = facets + .iter() + .map(|facet| label("Facet", &facet.name, facet.active())) + .chain( + automations + .iter() + .map(|automation| label("Automation", &automation.name, automation.active())), + ) + .collect::>(); + let defaults = facets + .iter() + .map(FacetTemplate::active) + .chain(automations.iter().map(AutomationTemplate::active)) + .collect::>(); + let term = + ui::prompt_term().ok_or_else(|| anyhow::anyhow!("interactive mode requires a TTY"))?; + let selected = MultiSelect::with_theme(&ColorfulTheme::default()) + .with_prompt("Select facets and Loop automations to include") + .items(&labels) + .defaults(&defaults) + .report(false) + .interact_on(&term) + .context("failed to select active observability resources")?; + Ok(filter_resources(facets, automations, &selected)) +} + +fn label(kind: &str, name: &str, active: bool) -> String { + format!( + "{kind:<12}{name}{}", + if active { "" } else { " (inactive)" } + ) +} + +fn filter_resources( + facets: Vec, + automations: Vec, + selected: &[usize], +) -> (Vec, Vec) { + let facet_count = facets.len(); + let selected = selected.iter().copied().collect::>(); + let facets = facets + .into_iter() + .enumerate() + .filter_map(|(index, facet)| selected.contains(&index).then_some(facet)) + .collect(); + let automations = automations + .into_iter() + .enumerate() + .filter_map(|(index, automation)| { + selected + .contains(&(facet_count + index)) + .then_some(automation) + }) + .collect(); + (facets, automations) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::active_observability_template::template::{KIND, SCHEMA_VERSION}; + + fn template() -> ActiveObservabilityTemplate { + serde_json::from_value(json!({ + "kind": KIND, + "schema_version": SCHEMA_VERSION, + "facets": [{ + "name": "Test facet", + "slug": "test-facet", + "function_data": {"type": "facet", "prompt": "Classify"} + }], + "automations": [{ + "name": "Test Loop", + "config": {"event_type": "windowed", "window": {}, "loop": {}} + }] + })) + .unwrap() + } + + #[test] + fn active_observability_stdout_is_only_pretty_json() { + let text = serialize_stdout(&template()).expect("serialize stdout"); + let value: serde_json::Value = serde_json::from_str(&text).expect("clean JSON"); + assert_eq!(value["kind"], KIND); + assert!(!text.contains("Pulled active observability")); + } + + #[test] + fn active_observability_selection_filters_both_resource_types() { + let template = template(); + let (facets, automations) = filter_resources(template.facets, template.automations, &[1]); + assert!(facets.is_empty()); + assert_eq!(automations.len(), 1); + } +} diff --git a/src/active_observability_template/push.rs b/src/active_observability_template/push.rs new file mode 100644 index 00000000..d1c9086f --- /dev/null +++ b/src/active_observability_template/push.rs @@ -0,0 +1,732 @@ +use std::collections::{BTreeMap, HashMap}; + +use anyhow::{anyhow, bail, Context, Result}; +use serde::Serialize; +use serde_json::{json, Value}; + +use crate::{ + functions::api::{create_function, replace_function, Function}, + http::ApiClient, + topics::api::{ + create_project_automation, patch_project_automation, replace_project_automation, + seed_new_topic_automation_cursors, ProjectAutomation, + }, +}; + +use super::template::{ + add_topics_functions, default_topics_config, embedding_model, is_loop_config, is_topics, + loop_config_for_target, new_topic_map_request, reconciled_topic_map_request, + saved_preprocessor_slug, topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, + AutomationTemplate, FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, +}; + +#[derive(Debug)] +pub(crate) struct Snapshot { + pub functions: Vec, + pub automations: Vec, +} + +#[derive(Debug)] +pub(crate) struct MutationPlan { + preprocessors: Vec>, + facets: Vec, + topics: BTreeMap, + loops: Vec, + function_ids: HashMap, +} + +#[derive(Debug)] +struct FunctionMutation { + template: T, + existing: Option, +} + +#[derive(Debug)] +struct FacetMutation { + template: FacetTemplate, + existing: Option, + topic_map: Option, + topics_key: String, +} + +#[derive(Debug)] +struct LoopMutation { + template: AutomationTemplate, + existing: Option, +} + +#[derive(Debug, Clone)] +enum TopicsTarget { + Existing(ProjectAutomation), + New(String), +} + +#[derive(Debug)] +struct TopicsMutation { + target: TopicsTarget, + embedding_model: String, + config: Value, +} + +#[derive(Debug, Serialize)] +pub(crate) struct PushResult { + pub facets: Vec, + pub automations: Vec, +} + +#[derive(Debug, serde::Serialize)] +pub(crate) struct PushedResource { + pub id: String, + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub slug: Option, +} + +pub(crate) fn plan( + template: &ActiveObservabilityTemplate, + snapshot: Snapshot, + topics_override: Option<&str>, + force: bool, +) -> Result { + let functions_by_slug = unique_functions_by_slug(&snapshot.functions)?; + let functions_by_id = snapshot + .functions + .iter() + .map(|function| (function.id.as_str(), function)) + .collect::>(); + let automations_by_name = unique_automations_by_name(&snapshot.automations)?; + + let mut preprocessor_templates = BTreeMap::new(); + for facet in &template.facets { + if let Some(preprocessor) = &facet.preprocessor { + preprocessor_templates + .entry(preprocessor.slug.clone()) + .or_insert_with(|| preprocessor.clone()); + } + } + let preprocessors = preprocessor_templates + .into_values() + .map(|template| { + let existing = checked_function( + functions_by_slug.get(template.slug.as_str()).copied(), + &template.slug, + "preprocessor", + "preprocessor", + )?; + conflict(existing, "preprocessor", &template.slug, force)?; + Ok(FunctionMutation { + template, + existing: existing.cloned(), + }) + }) + .collect::>>()?; + + let topics_automations = snapshot + .automations + .iter() + .filter(|automation| is_topics(&automation.config)) + .collect::>(); + let mut topics = BTreeMap::::new(); + let mut facets = Vec::with_capacity(template.facets.len()); + + for facet in &template.facets { + let existing = checked_function( + functions_by_slug.get(facet.slug.as_str()).copied(), + &facet.slug, + "facet", + "facet", + )?; + conflict(existing, "facet", &facet.slug, force)?; + + if let Some(slug) = saved_preprocessor_slug(&facet.function_data)? { + let target = functions_by_slug.get(slug).copied(); + if facet.preprocessor.is_none() && target.is_none() { + bail!( + "facet '{}' references preprocessor '{slug}', but it is not bundled or present in the target project", + facet.name + ); + } + if let Some(target) = target { + checked_function(Some(target), slug, "preprocessor", "preprocessor")?; + } + } + + let map_slug = topic_map_slug(&facet.slug); + let topic_map = checked_function( + functions_by_slug.get(map_slug.as_str()).copied(), + &map_slug, + "classifier topic map", + "classifier", + )?; + if let Some(topic_map) = topic_map { + if topic_map + .function_data + .as_ref() + .and_then(|data| data.get("type")) + .and_then(Value::as_str) + != Some("topic_map") + { + bail!( + "function slug '{map_slug}' is occupied by a classifier that is not a topic map" + ); + } + } + conflict(topic_map, "topic map", &map_slug, force)?; + + let target = resolve_topics_target( + facet, + topics_override, + &snapshot.automations, + &topics_automations, + &automations_by_name, + )?; + let key = target.key(); + let model = match &target { + TopicsTarget::Existing(automation) => embedding_model(automation, &functions_by_id), + TopicsTarget::New(_) => super::template::DEFAULT_EMBEDDING_MODEL.to_string(), + }; + // Validate destination-owned Topics configuration before any mutation begins. + let config = match &target { + TopicsTarget::Existing(automation) => add_topics_functions(&automation.config, &[])?, + TopicsTarget::New(_) => default_topics_config(), + }; + topics.entry(key.clone()).or_insert(TopicsMutation { + target, + embedding_model: model, + config, + }); + facets.push(FacetMutation { + template: facet.clone(), + existing: existing.cloned(), + topic_map: topic_map.cloned(), + topics_key: key, + }); + } + + let loops = template + .automations + .iter() + .map(|template| { + let existing = automations_by_name.get(template.name.as_str()).copied(); + if let Some(existing) = existing { + if !is_loop_config(&existing.config) { + bail!( + "automation '{}' already exists but is not a Loop automation", + template.name + ); + } + if !force { + bail!( + "Loop automation '{}' already exists; use --force to replace it", + template.name + ); + } + } + if topics.values().any(|topics| { + matches!(&topics.target, TopicsTarget::New(name) if name == &template.name) + }) { + bail!( + "template would create both a Topics and Loop automation named '{}'", + template.name + ); + } + Ok(LoopMutation { + template: template.clone(), + existing: existing.cloned(), + }) + }) + .collect::>>()?; + + Ok(MutationPlan { + preprocessors, + facets, + topics, + loops, + function_ids: snapshot + .functions + .into_iter() + .map(|function| (function.slug, function.id)) + .collect(), + }) +} + +fn unique_functions_by_slug(functions: &[Function]) -> Result> { + let mut by_slug = HashMap::new(); + for function in functions { + if by_slug.insert(function.slug.as_str(), function).is_some() { + bail!("multiple target functions have slug '{}'", function.slug); + } + } + Ok(by_slug) +} + +fn unique_automations_by_name( + automations: &[ProjectAutomation], +) -> Result> { + let mut by_name = HashMap::new(); + for automation in automations { + if by_name + .insert(automation.name.as_str(), automation) + .is_some() + { + bail!( + "multiple target automations are named '{}'", + automation.name + ); + } + } + Ok(by_name) +} + +fn checked_function<'a>( + function: Option<&'a Function>, + slug: &str, + label: &str, + expected_type: &str, +) -> Result> { + if let Some(function) = function { + if function.function_type.as_deref() != Some(expected_type) { + bail!( + "function slug '{slug}' is occupied by a '{}' function, not a {label}", + function.function_type.as_deref().unwrap_or("unknown") + ); + } + } + Ok(function) +} + +fn conflict(existing: Option<&Function>, label: &str, slug: &str, force: bool) -> Result<()> { + if existing.is_some() && !force { + bail!("{label} with slug '{slug}' already exists; use --force to replace it"); + } + Ok(()) +} + +fn resolve_topics_target( + facet: &FacetTemplate, + topics_override: Option<&str>, + all_automations: &[ProjectAutomation], + topics_automations: &[&ProjectAutomation], + by_name: &HashMap<&str, &ProjectAutomation>, +) -> Result { + if let Some(selector) = topics_override { + if selector.trim().is_empty() { + bail!("--topics-automation must not be empty"); + } + let matches = all_automations + .iter() + .filter(|automation| automation.id == selector || automation.name == selector) + .collect::>(); + let automation = match matches.as_slice() { + [] => bail!( + "Topics automation '{selector}' was not found; use an exact name or ID with --topics-automation" + ), + [automation] => *automation, + _ => bail!( + "--topics-automation '{selector}' is ambiguous; use the exact automation ID" + ), + }; + if !is_topics(&automation.config) { + bail!("automation '{selector}' is not a Topics automation"); + } + return Ok(TopicsTarget::Existing(automation.clone())); + } + + if let Some(name) = facet.topics_automation.as_deref() { + return match by_name.get(name).copied() { + Some(automation) if is_topics(&automation.config) => { + Ok(TopicsTarget::Existing(automation.clone())) + } + Some(_) => bail!( + "automation '{name}' exists but is not a Topics automation; use a different mapping" + ), + None => Ok(TopicsTarget::New(name.to_string())), + }; + } + + match topics_automations { + [automation] => Ok(TopicsTarget::Existing((*automation).clone())), + [] => bail!( + "no Topics automation can be inferred for facet '{}'; use --topics-automation ", + facet.name + ), + _ => bail!( + "multiple Topics automations exist for facet '{}'; use --topics-automation ", + facet.name + ), + } +} + +impl TopicsTarget { + fn key(&self) -> String { + match self { + Self::Existing(automation) => format!("id:{}", automation.id), + Self::New(name) => format!("new:{name}"), + } + } +} + +pub(crate) async fn execute( + client: &ApiClient, + project_id: &str, + mut plan: MutationPlan, +) -> Result { + for mutation in &plan.preprocessors { + let request = mutation.template.request(project_id, "preprocessor"); + let pushed = upsert_function(client, &request, mutation.existing.is_some()) + .await + .with_context(|| format!("failed to push preprocessor '{}'", mutation.template.slug))?; + plan.function_ids.insert(pushed.slug, pushed.id); + } + + let mut topic_functions: BTreeMap> = BTreeMap::new(); + let mut pushed_facets = Vec::with_capacity(plan.facets.len()); + for mutation in &plan.facets { + let preprocessor_id = saved_preprocessor_slug(&mutation.template.function_data)? + .map(|slug| { + plan.function_ids + .get(slug) + .map(String::as_str) + .ok_or_else(|| { + anyhow!("preprocessor '{slug}' disappeared from the mutation plan") + }) + }) + .transpose()?; + let function_data = + with_preprocessor_id(&mutation.template.function_data, preprocessor_id)?; + let request = mutation.template.request(project_id, &function_data); + let facet = upsert_function(client, &request, mutation.existing.is_some()) + .await + .with_context(|| format!("failed to push facet '{}'", mutation.template.slug))?; + + let topic_map = match &mutation.topic_map { + Some(existing) => { + let request = + reconciled_topic_map_request(existing, &mutation.template, &facet.id)?; + replace_function(client, &request) + .await + .with_context(|| format!("failed to reconcile topic map '{}'", existing.slug))? + } + None => { + let model = &plan + .topics + .get(&mutation.topics_key) + .expect("facet Topics plan exists") + .embedding_model; + let request = + new_topic_map_request(project_id, &mutation.template, &facet.id, model); + create_function(client, &request).await.with_context(|| { + format!( + "failed to create topic map '{}'", + topic_map_slug(&mutation.template.slug) + ) + })? + } + }; + topic_functions + .entry(mutation.topics_key.clone()) + .or_default() + .push((facet.id.clone(), topic_map.id)); + pushed_facets.push(PushedResource { + id: facet.id, + name: facet.name, + slug: Some(facet.slug), + }); + } + + for (key, mutation) in &plan.topics { + let pairs = topic_functions + .get(key) + .map(Vec::as_slice) + .unwrap_or_default(); + let config = add_topics_functions(&mutation.config, pairs)?; + match &mutation.target { + TopicsTarget::Existing(automation) => { + let body = json!({ + "name": automation.name, + "description": automation.description, + "config": config, + }); + patch_project_automation(client, &automation.id, &body) + .await + .with_context(|| { + format!("failed to update Topics automation '{}'", automation.name) + })?; + } + TopicsTarget::New(name) => { + let body = json!({ + "project_id": project_id, + "name": name, + "description": DEFAULT_TOPICS_DESCRIPTION, + "config": config, + }); + let created = create_project_automation(client, &body) + .await + .with_context(|| format!("failed to create Topics automation '{name}'"))?; + seed_new_topic_automation_cursors(client, project_id, &created) + .await + .with_context(|| format!("failed to seed Topics automation '{name}'"))?; + } + } + } + + let mut pushed_loops = Vec::with_capacity(plan.loops.len()); + for mutation in &plan.loops { + let existing_config = mutation.existing.as_ref().map(|row| &row.config); + let config = loop_config_for_target(&mutation.template.config, existing_config)?; + let pushed = if mutation.existing.is_some() { + let body = json!({ + "project_id": project_id, + "name": mutation.template.name, + "description": mutation.template.description, + "config": config, + }); + replace_project_automation(client, &body) + .await + .with_context(|| { + format!( + "failed to replace Loop automation '{}'", + mutation.template.name + ) + })? + } else { + let body = json!({ + "project_id": project_id, + "name": mutation.template.name, + "description": mutation.template.description, + "config": config, + }); + create_project_automation(client, &body) + .await + .with_context(|| { + format!( + "failed to create Loop automation '{}'", + mutation.template.name + ) + })? + }; + pushed_loops.push(PushedResource { + id: pushed.id, + name: pushed.name, + slug: None, + }); + } + + Ok(PushResult { + facets: pushed_facets, + automations: pushed_loops, + }) +} + +async fn upsert_function(client: &ApiClient, request: &Value, replace: bool) -> Result { + if replace { + replace_function(client, request).await + } else { + create_function(client, request).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::active_observability_template::template::{validate, KIND, SCHEMA_VERSION}; + + fn facet(topics: Option<&str>) -> FacetTemplate { + FacetTemplate { + name: "Test facet".to_string(), + slug: "test-facet".to_string(), + topics_automation: topics.map(str::to_string), + description: None, + preprocessor: None, + function_data: json!({"type": "facet", "prompt": "Classify this trace"}), + prompt_data: None, + tags: None, + function_schema: None, + } + } + + fn template(facet: FacetTemplate) -> ActiveObservabilityTemplate { + ActiveObservabilityTemplate { + kind: KIND.to_string(), + schema_version: SCHEMA_VERSION, + facets: vec![facet], + automations: Vec::new(), + } + } + + fn automation(id: &str, name: &str, event_type: &str) -> ProjectAutomation { + ProjectAutomation { + id: id.to_string(), + project_id: "test-project-id".to_string(), + name: name.to_string(), + description: None, + config: json!({ + "event_type": event_type, + "facet_functions": [], + "topic_map_functions": [] + }), + } + } + + fn existing_function(slug: &str, function_type: &str, data_type: &str) -> Function { + Function { + id: format!("fn-{slug}"), + name: slug.to_string(), + slug: slug.to_string(), + project_id: "test-project-id".to_string(), + description: None, + function_type: Some(function_type.to_string()), + prompt_data: None, + function_data: Some(json!({"type": data_type})), + tags: None, + function_schema: None, + metadata: None, + created: None, + _xact_id: None, + } + } + + #[test] + fn active_observability_plans_topics_selection_rules() { + let only = automation("auto-topics", "Topics", "topic"); + let inferred = plan( + &template(facet(None)), + Snapshot { + functions: vec![], + automations: vec![only.clone()], + }, + None, + false, + ) + .expect("single Topics automation"); + assert!(inferred.topics.contains_key("id:auto-topics")); + + let missing = plan( + &template(facet(None)), + Snapshot { + functions: vec![], + automations: vec![], + }, + None, + false, + ) + .expect_err("missing selector"); + assert!(missing.to_string().contains("--topics-automation")); + + let multiple = plan( + &template(facet(None)), + Snapshot { + functions: vec![], + automations: vec![only.clone(), automation("auto-other", "Other", "topic")], + }, + None, + false, + ) + .expect_err("ambiguous selector"); + assert!(multiple.to_string().contains("multiple Topics")); + + let selected = plan( + &template(facet(None)), + Snapshot { + functions: vec![], + automations: vec![only, automation("auto-other", "Other", "topic")], + }, + Some("auto-other"), + false, + ) + .expect("CLI override"); + assert!(selected.topics.contains_key("id:auto-other")); + } + + #[test] + fn active_observability_named_topics_mapping_can_plan_one_creation() { + let mut source = template(facet(Some("Synthetic Topics"))); + source.facets.push(FacetTemplate { + slug: "second-facet".to_string(), + name: "Second facet".to_string(), + ..facet(Some("Synthetic Topics")) + }); + validate(&source).expect("valid template"); + let plan = plan( + &source, + Snapshot { + functions: vec![], + automations: vec![], + }, + None, + false, + ) + .expect("one new destination"); + assert_eq!(plan.topics.len(), 1); + assert_eq!(plan.facets.len(), 2); + } + + #[test] + fn active_observability_preflight_rejects_type_and_no_force_conflicts() { + let wrong = plan( + &template(facet(Some("Topics"))), + Snapshot { + functions: vec![existing_function("test-facet", "tool", "code")], + automations: vec![automation("auto-topics", "Topics", "topic")], + }, + None, + true, + ) + .expect_err("wrong type"); + assert!(wrong.to_string().contains("not a facet")); + + let no_force = plan( + &template(facet(Some("Topics"))), + Snapshot { + functions: vec![existing_function("test-facet", "facet", "facet")], + automations: vec![automation("auto-topics", "Topics", "topic")], + }, + None, + false, + ) + .expect_err("no force conflict"); + assert!(no_force.to_string().contains("--force")); + + let wrong_loop = plan( + &ActiveObservabilityTemplate { + kind: KIND.to_string(), + schema_version: SCHEMA_VERSION, + facets: Vec::new(), + automations: vec![AutomationTemplate { + name: "Test automation".to_string(), + description: None, + config: json!({"event_type": "windowed", "window": {}, "loop": {}}), + }], + }, + Snapshot { + functions: vec![], + automations: vec![automation("auto-test", "Test automation", "topic")], + }, + None, + true, + ) + .expect_err("wrong automation type"); + assert!(wrong_loop.to_string().contains("not a Loop")); + } + + #[test] + fn active_observability_preflight_rejects_malformed_topics_config() { + let mut topics = automation("auto-topics", "Topics", "topic"); + topics.config["facet_functions"] = json!("not-an-array"); + + let error = plan( + &template(facet(Some("Topics"))), + Snapshot { + functions: vec![], + automations: vec![topics], + }, + None, + false, + ) + .expect_err("malformed Topics config"); + + assert!(error + .to_string() + .contains("Topics automation facet_functions must be an array")); + } +} diff --git a/src/active_observability_template/template.rs b/src/active_observability_template/template.rs new file mode 100644 index 00000000..c112604a --- /dev/null +++ b/src/active_observability_template/template.rs @@ -0,0 +1,832 @@ +use std::collections::{HashMap, HashSet}; + +use anyhow::{anyhow, bail, Result}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Map, Value}; + +use crate::{functions::api::Function, topics::api::ProjectAutomation}; + +pub(crate) const KIND: &str = "active_observability_template"; +pub(crate) const SCHEMA_VERSION: u32 = 1; +pub(crate) const DEFAULT_EMBEDDING_MODEL: &str = "brain-embedding-1"; +pub(crate) const DEFAULT_TOPICS_DESCRIPTION: &str = + "Automatically extract facets and classify logs using topic maps"; + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +pub(crate) struct ActiveObservabilityTemplate { + pub kind: String, + pub schema_version: u32, + #[serde(default)] + pub facets: Vec, + #[serde(default)] + pub automations: Vec, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +pub(crate) struct FacetTemplate { + pub name: String, + pub slug: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub topics_automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub preprocessor: Option, + pub function_data: Value, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prompt_data: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tags: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub function_schema: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +pub(crate) struct PortableFunction { + pub name: String, + pub slug: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + pub function_data: Value, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prompt_data: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tags: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub function_schema: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +pub(crate) struct AutomationTemplate { + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + pub config: Value, +} + +pub(crate) fn from_remote( + functions: &[Function], + automations: &[ProjectAutomation], +) -> Result { + let by_id = functions + .iter() + .map(|function| (function.id.as_str(), function)) + .collect::>(); + let facets = functions + .iter() + .filter(|function| function.function_type.as_deref() == Some("facet")) + .collect::>(); + let topics = topics_by_facet(&facets, &by_id, automations)?; + + let mut facet_templates = facets + .into_iter() + .map(|facet| facet_from_remote(facet, topics.get(&facet.id).cloned(), &by_id)) + .collect::>>()?; + facet_templates.sort_by(|a, b| a.name.cmp(&b.name).then(a.slug.cmp(&b.slug))); + + let mut loop_templates = automations + .iter() + .filter(|automation| is_loop_config(&automation.config)) + .map(|automation| { + let mut config = object(&automation.config, "Loop automation config")?.clone(); + config.remove("actions"); + Ok(AutomationTemplate { + name: automation.name.clone(), + description: automation.description.clone(), + config: Value::Object(config), + }) + }) + .collect::>>()?; + loop_templates.sort_by(|a, b| a.name.cmp(&b.name)); + + Ok(ActiveObservabilityTemplate { + kind: KIND.to_string(), + schema_version: SCHEMA_VERSION, + facets: facet_templates, + automations: loop_templates, + }) +} + +fn facet_from_remote( + facet: &Function, + topics_automation: Option, + by_id: &HashMap<&str, &Function>, +) -> Result { + let mut function_data = facet + .function_data + .clone() + .ok_or_else(|| anyhow!("facet '{}' is missing function_data", facet.name))?; + if function_data.get("type").and_then(Value::as_str) != Some("facet") { + bail!( + "function '{}' has facet type but non-facet function_data", + facet.name + ); + } + + let preprocessor = saved_preprocessor(&mut function_data, by_id)?; + Ok(FacetTemplate { + name: facet.name.clone(), + slug: facet.slug.clone(), + topics_automation, + description: facet.description.clone(), + preprocessor, + function_data, + prompt_data: facet.prompt_data.clone(), + tags: facet.tags.clone(), + function_schema: facet.function_schema.clone(), + }) +} + +fn saved_preprocessor( + function_data: &mut Value, + by_id: &HashMap<&str, &Function>, +) -> Result> { + let Some(reference) = function_data + .get_mut("preprocessor") + .and_then(Value::as_object_mut) + .filter(|reference| reference.get("type").and_then(Value::as_str) == Some("function")) + else { + return Ok(None); + }; + let id = reference + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("saved facet preprocessor reference is missing its function id"))?; + let function = by_id + .get(id) + .ok_or_else(|| anyhow!("facet references missing preprocessor function '{id}'"))?; + if function.function_type.as_deref() != Some("preprocessor") { + bail!("facet preprocessor reference '{id}' is not a preprocessor function"); + } + let portable = PortableFunction::from_remote(function)?; + *reference = Map::from_iter([ + ("type".to_string(), Value::String("function".to_string())), + ("slug".to_string(), Value::String(portable.slug.clone())), + ]); + Ok(Some(portable)) +} + +fn topics_by_facet( + facets: &[&Function], + by_id: &HashMap<&str, &Function>, + automations: &[ProjectAutomation], +) -> Result> { + let mut names: HashMap> = HashMap::new(); + for automation in automations + .iter() + .filter(|automation| is_topics(&automation.config)) + { + for id in topic_map_ids(&automation.config) { + let Some(topic_map) = by_id.get(id) else { + continue; + }; + if topic_map.function_type.as_deref() != Some("classifier") + || topic_map + .function_data + .as_ref() + .and_then(|data| data.get("type")) + .and_then(Value::as_str) + != Some("topic_map") + { + continue; + } + for facet in facets + .iter() + .copied() + .filter(|facet| topic_map_matches(topic_map, facet)) + { + names + .entry(facet.id.clone()) + .or_default() + .insert(automation.name.clone()); + } + } + } + + names + .into_iter() + .map(|(facet_id, names)| { + if names.len() != 1 { + let mut names = names.into_iter().collect::>(); + names.sort(); + bail!( + "facet '{facet_id}' belongs to multiple Topics automations ({}); use one Topics destination per facet", + names.join(", ") + ); + } + Ok((facet_id, names.into_iter().next().expect("one name"))) + }) + .collect() +} + +fn topic_map_matches(topic_map: &Function, facet: &Function) -> bool { + let Some(data) = topic_map.function_data.as_ref() else { + return false; + }; + match data + .get("source_facet_function") + .and_then(Value::as_object) + .filter(|reference| reference.get("type").and_then(Value::as_str) == Some("function")) + .and_then(|reference| reference.get("id")) + .and_then(Value::as_str) + { + Some(id) => id == facet.id, + None => data + .get("source_facet") + .and_then(Value::as_str) + .is_some_and(|source| source == facet.name || source == facet.slug), + } +} + +pub(crate) fn validate(template: &ActiveObservabilityTemplate) -> Result<()> { + if template.kind != KIND { + bail!("template kind must be '{KIND}'"); + } + if template.schema_version != SCHEMA_VERSION { + bail!( + "unsupported template schema version {}; supported version is {SCHEMA_VERSION}", + template.schema_version + ); + } + + let mut slugs = HashMap::::new(); + let mut preprocessors = HashMap::::new(); + for facet in &template.facets { + require_text(&facet.name, "facet name")?; + require_text(&facet.slug, "facet slug")?; + if facet.function_data.get("type").and_then(Value::as_str) != Some("facet") { + bail!("facet '{}' function_data.type must be 'facet'", facet.name); + } + reserve_slug(&mut slugs, &facet.slug, "facet")?; + reserve_slug( + &mut slugs, + &topic_map_slug(&facet.slug), + "generated topic map", + )?; + if let Some(name) = facet.topics_automation.as_deref() { + require_text(name, "Topics automation name")?; + } + + let saved_slug = saved_preprocessor_slug(&facet.function_data)?; + match (&facet.preprocessor, saved_slug) { + (Some(preprocessor), Some(slug)) => { + preprocessor.validate()?; + if preprocessor.slug != slug { + bail!( + "facet '{}' bundles preprocessor '{}' but references '{}'", + facet.name, + preprocessor.slug, + slug + ); + } + match preprocessors.get(slug) { + Some(existing) if **existing != *preprocessor => { + bail!("template contains conflicting preprocessors with slug '{slug}'") + } + Some(_) => {} + None => { + reserve_slug(&mut slugs, slug, "bundled preprocessor")?; + preprocessors.insert(slug.to_string(), preprocessor); + } + } + } + (Some(_), None) => bail!( + "facet '{}' bundles a preprocessor without a saved preprocessor reference", + facet.name + ), + _ => {} + } + } + + let mut automation_names = HashSet::new(); + for automation in &template.automations { + require_text(&automation.name, "automation name")?; + if !automation_names.insert(automation.name.as_str()) { + bail!( + "template contains duplicate automation name '{}'", + automation.name + ); + } + if !is_loop_config(&automation.config) { + bail!( + "automation '{}' is not a Loop automation (expected windowed config with loop)", + automation.name + ); + } + } + Ok(()) +} + +fn reserve_slug( + slugs: &mut HashMap, + slug: &str, + kind: &'static str, +) -> Result<()> { + if let Some(previous) = slugs.insert(slug.to_string(), kind) { + if previous == "facet" && kind == "facet" { + bail!("template contains duplicate facet slug '{slug}'"); + } + bail!("template uses function slug '{slug}' for both {previous} and {kind}"); + } + Ok(()) +} + +fn require_text(value: &str, label: &str) -> Result<()> { + if value.trim().is_empty() { + bail!("{label} must not be empty"); + } + Ok(()) +} + +impl PortableFunction { + fn from_remote(function: &Function) -> Result { + Ok(Self { + name: function.name.clone(), + slug: function.slug.clone(), + description: function.description.clone(), + function_data: function + .function_data + .clone() + .ok_or_else(|| anyhow!("function '{}' is missing function_data", function.name))?, + prompt_data: function.prompt_data.clone(), + tags: function.tags.clone(), + function_schema: function.function_schema.clone(), + }) + } + + fn validate(&self) -> Result<()> { + require_text(&self.name, "preprocessor name")?; + require_text(&self.slug, "preprocessor slug")?; + if !self.function_data.is_object() { + bail!( + "preprocessor '{}' function_data must be an object", + self.name + ); + } + Ok(()) + } + + pub(crate) fn request(&self, project_id: &str, function_type: &str) -> Value { + portable_function_request( + project_id, + &self.name, + &self.slug, + self.description.as_ref(), + function_type, + &self.function_data, + self.prompt_data.as_ref(), + self.tags.as_ref(), + self.function_schema.as_ref(), + ) + } +} + +impl FacetTemplate { + pub(crate) fn request(&self, project_id: &str, function_data: &Value) -> Value { + portable_function_request( + project_id, + &self.name, + &self.slug, + self.description.as_ref(), + "facet", + function_data, + self.prompt_data.as_ref(), + self.tags.as_ref(), + self.function_schema.as_ref(), + ) + } + + pub(crate) fn active(&self) -> bool { + self.topics_automation.is_some() + } +} + +fn portable_function_request( + project_id: &str, + name: &str, + slug: &str, + description: Option<&String>, + function_type: &str, + function_data: &Value, + prompt_data: Option<&Value>, + tags: Option<&Vec>, + function_schema: Option<&Value>, +) -> Value { + json!({ + "project_id": project_id, + "name": name, + "slug": slug, + "description": description, + "function_type": function_type, + "function_data": function_data, + "prompt_data": prompt_data, + "tags": tags, + "function_schema": function_schema, + }) +} + +pub(crate) fn saved_preprocessor_slug(function_data: &Value) -> Result> { + let Some(reference) = function_data + .get("preprocessor") + .and_then(Value::as_object) + .filter(|reference| reference.get("type").and_then(Value::as_str) == Some("function")) + else { + return Ok(None); + }; + if reference.contains_key("id") { + bail!("portable saved preprocessor reference must use 'slug', not source-project 'id'"); + } + reference + .get("slug") + .and_then(Value::as_str) + .filter(|slug| !slug.trim().is_empty()) + .map(Some) + .ok_or_else(|| anyhow!("portable saved preprocessor reference is missing its slug")) +} + +pub(crate) fn with_preprocessor_id(function_data: &Value, id: Option<&str>) -> Result { + let mut data = object(function_data, "facet function_data")?.clone(); + if let Some(id) = id { + data.insert( + "preprocessor".to_string(), + json!({"type": "function", "id": id}), + ); + } + Ok(Value::Object(data)) +} + +pub(crate) fn topic_map_slug(facet_slug: &str) -> String { + format!("{facet_slug}-topic-map") +} + +pub(crate) fn new_topic_map_request( + project_id: &str, + facet: &FacetTemplate, + facet_id: &str, + embedding_model: &str, +) -> Value { + json!({ + "project_id": project_id, + "name": facet.name, + "slug": topic_map_slug(&facet.slug), + "description": facet.description, + "function_type": "classifier", + "function_data": { + "type": "topic_map", + "source_facet": facet.name, + "source_facet_function": {"type": "function", "id": facet_id}, + "embedding_model": embedding_model, + } + }) +} + +pub(crate) fn reconciled_topic_map_request( + existing: &Function, + facet: &FacetTemplate, + facet_id: &str, +) -> Result { + let mut data = object( + existing + .function_data + .as_ref() + .ok_or_else(|| anyhow!("topic map '{}' is missing function_data", existing.slug))?, + "topic map function_data", + )? + .clone(); + data.insert( + "source_facet".to_string(), + Value::String(facet.name.clone()), + ); + data.insert( + "source_facet_function".to_string(), + json!({"type": "function", "id": facet_id}), + ); + Ok(portable_function_request( + &existing.project_id, + &facet.name, + &existing.slug, + existing.description.as_ref(), + "classifier", + &Value::Object(data), + existing.prompt_data.as_ref(), + existing.tags.as_ref(), + existing.function_schema.as_ref(), + )) +} + +pub(crate) fn deduplicate_preprocessors(facets: &mut [FacetTemplate]) { + let mut included = HashSet::new(); + for facet in facets { + if facet + .preprocessor + .as_ref() + .is_some_and(|preprocessor| !included.insert(preprocessor.slug.clone())) + { + facet.preprocessor = None; + } + } +} + +pub(crate) fn is_topics(config: &Value) -> bool { + config.get("event_type").and_then(Value::as_str) == Some("topic") +} + +pub(crate) fn is_loop_config(config: &Value) -> bool { + config.get("event_type").and_then(Value::as_str) == Some("windowed") + && config.get("loop").and_then(Value::as_object).is_some() +} + +pub(crate) fn loop_config_for_target(template: &Value, existing: Option<&Value>) -> Result { + let mut config = object(template, "Loop automation config")?.clone(); + let actions = existing + .and_then(Value::as_object) + .and_then(|config| config.get("actions")) + .cloned() + .unwrap_or_else(|| Value::Array(Vec::new())); + config.insert("actions".to_string(), actions); + Ok(Value::Object(config)) +} + +pub(crate) fn default_topics_config() -> Value { + json!({ + "event_type": "topic", + "sampling_rate": 1.0, + "facet_functions": [], + "topic_map_functions": [], + "scope": {"type": "trace", "idle_seconds": 600}, + "rerun_seconds": 86400, + "relabel_overlap_seconds": 3600, + "backfill_time_range": "86400s", + }) +} + +pub(crate) fn add_topics_functions( + config: &Value, + functions: &[(String, String)], +) -> Result { + let mut config = object(config, "Topics automation config")?.clone(); + let facets = array_entry(&mut config, "facet_functions")?; + for (facet_id, _) in functions { + if !facets + .iter() + .any(|entry| function_ref_id(entry) == Some(facet_id.as_str())) + { + facets.push(json!({"type": "function", "id": facet_id})); + } + } + let topic_maps = array_entry(&mut config, "topic_map_functions")?; + for (_, topic_map_id) in functions { + if !topic_maps.iter().any(|entry| { + entry.get("function").and_then(function_ref_id) == Some(topic_map_id.as_str()) + }) { + topic_maps.push(json!({"function": {"type": "function", "id": topic_map_id}})); + } + } + Ok(Value::Object(config)) +} + +pub(crate) fn embedding_model( + automation: &ProjectAutomation, + functions_by_id: &HashMap<&str, &Function>, +) -> String { + topic_map_ids(&automation.config) + .filter_map(|id| functions_by_id.get(id)) + .filter_map(|function| function.function_data.as_ref()) + .filter_map(|data| data.get("embedding_model").and_then(Value::as_str)) + .find(|model| !model.trim().is_empty()) + .unwrap_or(DEFAULT_EMBEDDING_MODEL) + .to_string() +} + +fn topic_map_ids(config: &Value) -> impl Iterator { + config + .get("topic_map_functions") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|entry| entry.get("function")) + .filter_map(function_ref_id) +} + +fn function_ref_id(reference: &Value) -> Option<&str> { + (reference.get("type").and_then(Value::as_str) == Some("function")) + .then(|| reference.get("id").and_then(Value::as_str)) + .flatten() +} + +fn array_entry<'a>(config: &'a mut Map, name: &str) -> Result<&'a mut Vec> { + config + .entry(name.to_string()) + .or_insert_with(|| Value::Array(Vec::new())) + .as_array_mut() + .ok_or_else(|| anyhow!("Topics automation {name} must be an array")) +} + +fn object<'a>(value: &'a Value, label: &str) -> Result<&'a Map> { + value + .as_object() + .ok_or_else(|| anyhow!("{label} must be a JSON object")) +} + +impl AutomationTemplate { + pub(crate) fn active(&self) -> bool { + self.config.get("status").and_then(Value::as_str) != Some("paused") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn function(id: &str, slug: &str, function_type: &str, data: Value) -> Function { + Function { + id: id.to_string(), + name: slug.to_string(), + slug: slug.to_string(), + project_id: "test-project-id".to_string(), + description: None, + function_type: Some(function_type.to_string()), + prompt_data: None, + function_data: Some(data), + tags: None, + function_schema: None, + metadata: None, + created: None, + _xact_id: None, + } + } + + fn automation(name: &str, config: Value) -> ProjectAutomation { + ProjectAutomation { + id: format!("test-{name}-id"), + project_id: "test-project-id".to_string(), + name: name.to_string(), + description: None, + config, + } + } + + #[test] + fn active_observability_pull_is_portable_and_maps_topics() { + let functions = vec![ + function( + "fn-test-preprocessor", + "test-preprocessor", + "preprocessor", + json!({"type": "code", "data": {"type": "inline", "code": "return input"}}), + ), + function( + "fn-test-facet", + "test-facet", + "facet", + json!({ + "type": "facet", + "prompt": "Classify this trace", + "preprocessor": {"type": "function", "id": "fn-test-preprocessor"} + }), + ), + function( + "fn-test-topic-map", + "test-facet-topic-map", + "classifier", + json!({ + "type": "topic_map", + "source_facet": "legacy-name", + "source_facet_function": {"type": "function", "id": "fn-test-facet"}, + "embedding_model": "test-embedding-model" + }), + ), + ]; + let automations = vec![ + automation( + "Topics", + json!({ + "event_type": "topic", + "topic_map_functions": [{"function": {"type": "function", "id": "fn-test-topic-map"}}] + }), + ), + automation( + "Test Loop", + json!({ + "event_type": "windowed", + "window": {}, + "loop": {}, + "actions": [{"type": "webhook", "url": "https://example.invalid/hook"}] + }), + ), + ]; + + let template = from_remote(&functions, &automations).expect("portable template"); + let value = serde_json::to_value(&template).expect("serialize"); + + assert_eq!( + template.facets[0].topics_automation.as_deref(), + Some("Topics") + ); + assert_eq!( + template.facets[0].function_data["preprocessor"], + json!({"type": "function", "slug": "test-preprocessor"}) + ); + assert_eq!( + template.facets[0].preprocessor.as_ref().unwrap().slug, + "test-preprocessor" + ); + assert!(value["facets"][0].get("id").is_none()); + assert!(value["automations"][0]["config"].get("actions").is_none()); + } + + #[test] + fn active_observability_pull_bundles_shared_preprocessor_once() { + let shared = PortableFunction { + name: "Shared preprocessor".to_string(), + slug: "shared-preprocessor".to_string(), + description: None, + function_data: json!({"type": "code", "data": {"type": "inline"}}), + prompt_data: None, + tags: None, + function_schema: None, + }; + let facet = |name: &str, slug: &str| FacetTemplate { + name: name.to_string(), + slug: slug.to_string(), + topics_automation: Some("Topics".to_string()), + description: None, + preprocessor: Some(shared.clone()), + function_data: json!({ + "type": "facet", + "preprocessor": {"type": "function", "slug": "shared-preprocessor"} + }), + prompt_data: None, + tags: None, + function_schema: None, + }; + let mut facets = vec![ + facet("First facet", "first-facet"), + facet("Second facet", "second-facet"), + ]; + + deduplicate_preprocessors(&mut facets); + + assert!(facets[0].preprocessor.is_some()); + assert!(facets[1].preprocessor.is_none()); + assert!(facets.iter().all(|facet| { + facet.function_data["preprocessor"] + == json!({"type": "function", "slug": "shared-preprocessor"}) + })); + } + + #[test] + fn active_observability_force_conversions_preserve_customization_and_actions() { + let topic_map = function( + "fn-test-topic-map", + "test-facet-topic-map", + "classifier", + json!({ + "type": "topic_map", + "source_facet": "Old", + "embedding_model": "custom-model", + "generation_settings": {"algorithm": "kmeans"}, + "report_key": "remote-report" + }), + ); + let facet: FacetTemplate = serde_json::from_value(json!({ + "name": "Test facet", "slug": "test-facet", "function_data": {"type": "facet", "prompt": "Test"} + })).unwrap(); + let request = reconciled_topic_map_request(&topic_map, &facet, "fn-test-facet").unwrap(); + assert_eq!(request["function_data"]["embedding_model"], "custom-model"); + assert_eq!( + request["function_data"]["generation_settings"]["algorithm"], + "kmeans" + ); + assert_eq!(request["function_data"]["report_key"], "remote-report"); + assert_eq!(request["function_data"]["source_facet"], "Test facet"); + assert_eq!( + request["function_data"]["source_facet_function"], + json!({"type": "function", "id": "fn-test-facet"}) + ); + + let config = loop_config_for_target( + &json!({"event_type": "windowed", "window": {}, "loop": {}, "actions": ["source"]}), + Some(&json!({"event_type": "windowed", "loop": {}, "actions": ["target"]})), + ) + .unwrap(); + assert_eq!(config["actions"], json!(["target"])); + } + + #[test] + fn active_observability_topics_update_is_idempotent() { + let config = json!({ + "event_type": "topic", + "custom": {"keep": true}, + "facet_functions": [{"type": "function", "id": "fn-test-facet"}], + "topic_map_functions": [] + }); + let pairs = vec![("fn-test-facet".to_string(), "fn-test-topic-map".to_string())]; + let once = add_topics_functions(&config, &pairs).unwrap(); + let twice = add_topics_functions(&once, &pairs).unwrap(); + assert_eq!(once, twice); + assert_eq!(twice["custom"]["keep"], true); + assert_eq!(twice["facet_functions"].as_array().unwrap().len(), 1); + assert_eq!(twice["topic_map_functions"].as_array().unwrap().len(), 1); + } +} diff --git a/src/functions/api.rs b/src/functions/api.rs index 0e526cfd..958221c7 100644 --- a/src/functions/api.rs +++ b/src/functions/api.rs @@ -1,3 +1,5 @@ +use std::collections::HashSet; + use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -29,6 +31,8 @@ pub struct Function { #[serde(default)] pub tags: Option>, #[serde(default)] + pub function_schema: Option, + #[serde(default)] pub metadata: Option, #[serde(default)] pub created: Option, @@ -298,6 +302,47 @@ pub async fn list_functions_page( parse_function_list_page(raw) } +/// List a project's functions using the public API's cursor contract. +pub async fn list_all_functions(client: &ApiClient, project_id: &str) -> Result> { + let mut query = FunctionListQuery { + project_id: Some(project_id.to_string()), + ..Default::default() + }; + let mut functions = Vec::new(); + let mut cursors = HashSet::new(); + + loop { + let page = list_functions_page(client, &query).await?; + if query.snapshot.is_none() { + query.snapshot = page.snapshot; + } + functions.extend( + page.objects + .into_iter() + .map(serde_json::from_value) + .collect::, _>>() + .context("unexpected function response shape")?, + ); + let Some(cursor) = page.next_cursor else { + break; + }; + if !cursors.insert(cursor.clone()) { + anyhow::bail!("function pagination returned a repeated cursor"); + } + query.cursor = Some(cursor); + } + + Ok(functions) +} + +pub async fn create_function(client: &ApiClient, body: &Value) -> Result { + client.post("/v1/function", body).await +} + +pub async fn replace_function(client: &ApiClient, body: &Value) -> Result { + client.put("/v1/function", body).await +} + fn parse_function_list_page(raw: Value) -> Result { let objects = raw .get("objects") diff --git a/src/functions/mod.rs b/src/functions/mod.rs index 03c5301a..9117135a 100644 --- a/src/functions/mod.rs +++ b/src/functions/mod.rs @@ -1116,6 +1116,7 @@ mod tests { prompt_data: Some(serde_json::json!({"parser": {"choice": ["a", "b"]}})), function_data: Some(serde_json::json!({"type": "prompt"})), tags: None, + function_schema: None, metadata: None, created: None, _xact_id: None, @@ -1136,6 +1137,7 @@ mod tests { prompt_data: None, function_data: None, tags: None, + function_schema: None, metadata: None, created: None, _xact_id: None, @@ -1163,6 +1165,7 @@ mod tests { prompt_data: None, function_data: None, tags: None, + function_schema: None, metadata: None, created: None, _xact_id: None, diff --git a/src/http.rs b/src/http.rs index 8676d015..00cbacf4 100644 --- a/src/http.rs +++ b/src/http.rs @@ -233,6 +233,26 @@ impl ApiClient { parse_json_response(response, "POST", path).await } + pub async fn put(&self, path: &str, body: &B) -> Result { + let url = self.url(path); + let response = self + .http + .put(&url) + .bearer_auth(&self.api_key) + .json(body) + .send() + .await + .context("request failed")?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(HttpError { status, body }.into()); + } + + parse_json_response(response, "PUT", path).await + } + pub async fn patch( &self, path: &str, diff --git a/src/main.rs b/src/main.rs index 4060a8be..93fe9c16 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ use anyhow::{Context, Result}; use clap::{parser::ValueSource, ArgMatches, CommandFactory, FromArgMatches, Parser, Subcommand}; use std::ffi::{OsStr, OsString}; +mod active_observability_template; mod args; mod auth; #[allow(dead_code)] @@ -69,6 +70,7 @@ Core Projects & resources projects Manage projects + active-observability-template Pull and push portable active observability templates topics Inspect and control Topics automation prompts Manage prompts functions Manage functions (tools, scorers, and more) @@ -149,6 +151,10 @@ enum Commands { Eval(CLIArgs), /// Manage projects Projects(CLIArgs), + /// Pull and push facets and Loop automations as a portable template + ActiveObservabilityTemplate( + CLIArgs, + ), /// Inspect and control Topics automation Topics(CLIArgs), /// Manage datasets @@ -198,6 +204,7 @@ impl Commands { #[cfg(unix)] Commands::Eval(cmd) => &cmd.base, Commands::Projects(cmd) => &cmd.base, + Commands::ActiveObservabilityTemplate(cmd) => &cmd.base, Commands::Topics(cmd) => &cmd.base, Commands::Datasets(cmd) => &cmd.base, Commands::Environments(cmd) => &cmd.base, @@ -229,6 +236,7 @@ impl Commands { #[cfg(unix)] Commands::Eval(cmd) => &mut cmd.base, Commands::Projects(cmd) => &mut cmd.base, + Commands::ActiveObservabilityTemplate(cmd) => &mut cmd.base, Commands::Datasets(cmd) => &mut cmd.base, Commands::Environments(cmd) => &mut cmd.base, Commands::Topics(cmd) => &mut cmd.base, @@ -363,6 +371,9 @@ fn try_main() -> Result<()> { #[cfg(unix)] Commands::Eval(cmd) => eval::run(cmd.base, cmd.args).await?, Commands::Projects(cmd) => projects::run(cmd.base, cmd.args).await?, + Commands::ActiveObservabilityTemplate(cmd) => { + active_observability_template::run(cmd.base, cmd.args).await? + } Commands::Datasets(cmd) => datasets::run(cmd.base, cmd.args).await?, Commands::Environments(cmd) => environments::run(cmd.base, cmd.args).await?, Commands::Topics(cmd) => topics::run(cmd.base, cmd.args).await?, diff --git a/src/topics/api.rs b/src/topics/api.rs index 736276ee..30510a0c 100644 --- a/src/topics/api.rs +++ b/src/topics/api.rs @@ -8,6 +8,53 @@ use urlencoding::encode; use crate::{http::ApiClient, project_context::ProjectContext, utils::app_project_url}; +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +pub(crate) struct ProjectAutomation { + pub id: String, + pub project_id: String, + pub name: String, + #[serde(default)] + pub description: Option, + pub config: Value, +} + +#[derive(Debug, Deserialize)] +struct ListResponse { + objects: Vec, +} + +pub(crate) async fn list_project_automations( + client: &ApiClient, + project_id: &str, +) -> Result> { + let path = format!("/v1/project_automation?project_id={}", encode(project_id)); + let response: ListResponse = client.get(&path).await?; + Ok(response.objects) +} + +pub(crate) async fn create_project_automation( + client: &ApiClient, + body: &Value, +) -> Result { + client.post("/v1/project_automation", body).await +} + +pub(crate) async fn replace_project_automation( + client: &ApiClient, + body: &Value, +) -> Result { + client.put("/v1/project_automation", body).await +} + +pub(crate) async fn patch_project_automation( + client: &ApiClient, + automation_id: &str, + body: &Value, +) -> Result { + let path = format!("/v1/project_automation/{}", encode(automation_id)); + client.patch(&path, body).await +} + const DEFAULT_TOPIC_AUTOMATION_NAME: &str = "Topics"; const DEFAULT_TOPIC_AUTOMATION_DESCRIPTION: &str = "Automatically extract facets and classify logs using topic maps"; @@ -957,6 +1004,16 @@ async fn seed_topic_automation_cursors( }) } +pub(crate) async fn seed_new_topic_automation_cursors( + client: &ApiClient, + project_id: &str, + automation: &ProjectAutomation, +) -> Result<()> { + seed_topic_automation_cursors(client, project_id, &serde_json::to_value(automation)?, None) + .await?; + Ok(()) +} + fn filter_or_resolve_topic_automation_rows( rows: Vec, automation_id: Option<&str>, From dd93c8254b78939ae719e6acd42dc0f32634392e Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 26 Aug 2026 16:10:39 -0700 Subject: [PATCH 2/4] fix: preserve active observability wiring --- src/active_observability_template/pull.rs | 54 ++++++++++ src/active_observability_template/push.rs | 101 +++++++++++++++++- src/active_observability_template/template.rs | 38 +++++++ 3 files changed, 191 insertions(+), 2 deletions(-) diff --git a/src/active_observability_template/pull.rs b/src/active_observability_template/pull.rs index b6cc0173..cd795872 100644 --- a/src/active_observability_template/pull.rs +++ b/src/active_observability_template/pull.rs @@ -35,6 +35,9 @@ pub(crate) async fn run(base: BaseArgs, args: PullArgs) -> Result<()> { if !base.json && !base.no_input && ui::is_interactive() { (template.facets, template.automations) = select_resources(template.facets, template.automations)?; + } else { + (template.facets, template.automations) = + filter_active_resources(template.facets, template.automations); } // Selection happens first so a selected facet never loses its required definition. deduplicate_preprocessors(&mut template.facets); @@ -167,6 +170,19 @@ fn filter_resources( (facets, automations) } +fn filter_active_resources( + facets: Vec, + automations: Vec, +) -> (Vec, Vec) { + ( + facets.into_iter().filter(FacetTemplate::active).collect(), + automations + .into_iter() + .filter(AutomationTemplate::active) + .collect(), + ) +} + #[cfg(test)] mod tests { use serde_json::json; @@ -206,4 +222,42 @@ mod tests { assert!(facets.is_empty()); assert_eq!(automations.len(), 1); } + + #[test] + fn active_observability_noninteractive_pull_uses_active_defaults() { + let mut template = template(); + template.facets.push(FacetTemplate { + name: "Active facet".to_string(), + slug: "active-facet".to_string(), + topics_automation: Some("Synthetic Topics".to_string()), + ..template.facets[0].clone() + }); + template.automations.push(AutomationTemplate { + name: "Paused Loop".to_string(), + description: None, + config: json!({ + "event_type": "windowed", + "status": "paused", + "window": {}, + "loop": {} + }), + }); + + let (facets, automations) = filter_active_resources(template.facets, template.automations); + + assert_eq!( + facets + .iter() + .map(|facet| facet.name.as_str()) + .collect::>(), + ["Active facet"] + ); + assert_eq!( + automations + .iter() + .map(|automation| automation.name.as_str()) + .collect::>(), + ["Test Loop"] + ); + } } diff --git a/src/active_observability_template/push.rs b/src/active_observability_template/push.rs index d1c9086f..dc0e4d13 100644 --- a/src/active_observability_template/push.rs +++ b/src/active_observability_template/push.rs @@ -16,8 +16,9 @@ use crate::{ use super::template::{ add_topics_functions, default_topics_config, embedding_model, is_loop_config, is_topics, loop_config_for_target, new_topic_map_request, reconciled_topic_map_request, - saved_preprocessor_slug, topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, - AutomationTemplate, FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, + remove_topics_functions, saved_preprocessor_slug, topic_map_slug, with_preprocessor_id, + ActiveObservabilityTemplate, AutomationTemplate, FacetTemplate, PortableFunction, + DEFAULT_TOPICS_DESCRIPTION, }; #[derive(Debug)] @@ -195,6 +196,14 @@ pub(crate) fn plan( embedding_model: model, config, }); + plan_topics_removals( + &mut topics, + &topics_automations, + &key, + existing, + topic_map, + &functions_by_id, + )?; facets.push(FacetMutation { template: facet.clone(), existing: existing.cloned(), @@ -250,6 +259,53 @@ pub(crate) fn plan( }) } +fn plan_topics_removals( + topics: &mut BTreeMap, + automations: &[&ProjectAutomation], + selected_key: &str, + facet: Option<&Function>, + topic_map: Option<&Function>, + functions_by_id: &HashMap<&str, &Function>, +) -> Result<()> { + if facet.is_none() && topic_map.is_none() { + return Ok(()); + } + + for &automation in automations { + let key = TopicsTarget::Existing(automation.clone()).key(); + if key == selected_key { + continue; + } + let current_config = topics + .get(&key) + .map(|mutation| &mutation.config) + .unwrap_or(&automation.config); + let Some(config) = remove_topics_functions( + current_config, + facet.map(|facet| facet.id.as_str()), + topic_map.map(|topic_map| topic_map.id.as_str()), + )? + else { + continue; + }; + + match topics.get_mut(&key) { + Some(mutation) => mutation.config = config, + None => { + topics.insert( + key, + TopicsMutation { + target: TopicsTarget::Existing(automation.clone()), + embedding_model: embedding_model(automation, functions_by_id), + config, + }, + ); + } + } + } + Ok(()) +} + fn unique_functions_by_slug(functions: &[Function]) -> Result> { let mut by_slug = HashMap::new(); for function in functions { @@ -661,6 +717,47 @@ mod tests { assert_eq!(plan.facets.len(), 2); } + #[test] + fn active_observability_remap_removes_the_previous_topics_wiring() { + let mut previous = automation("auto-previous", "Previous Topics", "topic"); + previous.config["facet_functions"] = json!([ + {"type": "function", "id": "fn-test-facet"}, + {"type": "function", "id": "fn-unrelated-facet"} + ]); + previous.config["topic_map_functions"] = json!([ + {"function": {"type": "function", "id": "fn-test-facet-topic-map"}}, + {"function": {"type": "function", "id": "fn-unrelated-topic-map"}} + ]); + + let planned = plan( + &template(facet(Some("Destination Topics"))), + Snapshot { + functions: vec![ + existing_function("test-facet", "facet", "facet"), + existing_function("test-facet-topic-map", "classifier", "topic_map"), + ], + automations: vec![ + previous, + automation("auto-destination", "Destination Topics", "topic"), + ], + }, + None, + true, + ) + .expect("facet remap"); + + let previous = &planned.topics["id:auto-previous"].config; + assert_eq!( + previous["facet_functions"], + json!([{"type": "function", "id": "fn-unrelated-facet"}]) + ); + assert_eq!( + previous["topic_map_functions"], + json!([{"function": {"type": "function", "id": "fn-unrelated-topic-map"}}]) + ); + assert!(planned.topics.contains_key("id:auto-destination")); + } + #[test] fn active_observability_preflight_rejects_type_and_no_force_conflicts() { let wrong = plan( diff --git a/src/active_observability_template/template.rs b/src/active_observability_template/template.rs index c112604a..7319492c 100644 --- a/src/active_observability_template/template.rs +++ b/src/active_observability_template/template.rs @@ -585,6 +585,44 @@ pub(crate) fn add_topics_functions( Ok(Value::Object(config)) } +pub(crate) fn remove_topics_functions( + config: &Value, + facet_id: Option<&str>, + topic_map_id: Option<&str>, +) -> Result> { + let contains_facet = facet_id.is_some_and(|id| { + config + .get("facet_functions") + .and_then(Value::as_array) + .is_some_and(|facets| { + facets + .iter() + .any(|entry| function_ref_id(entry) == Some(id)) + }) + }); + let contains_topic_map = topic_map_id.is_some_and(|id| { + config + .get("topic_map_functions") + .and_then(Value::as_array) + .is_some_and(|topic_maps| { + topic_maps + .iter() + .any(|entry| entry.get("function").and_then(function_ref_id) == Some(id)) + }) + }); + if !contains_facet && !contains_topic_map { + return Ok(None); + } + + let mut config = object(config, "Topics automation config")?.clone(); + array_entry(&mut config, "facet_functions")? + .retain(|entry| facet_id.is_none_or(|id| function_ref_id(entry) != Some(id))); + array_entry(&mut config, "topic_map_functions")?.retain(|entry| { + topic_map_id.is_none_or(|id| entry.get("function").and_then(function_ref_id) != Some(id)) + }); + Ok(Some(Value::Object(config))) +} + pub(crate) fn embedding_model( automation: &ProjectAutomation, functions_by_id: &HashMap<&str, &Function>, From f0d76525af5028f7ea2fe537cd3d5a99b4bef36c Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 26 Aug 2026 16:21:45 -0700 Subject: [PATCH 3/4] refactor: group observability template commands --- src/main.rs | 64 +++++++++--------- .../mod.rs | 66 ++++++++++++------- .../pull.rs | 2 +- .../push.rs | 2 +- .../template.rs | 0 5 files changed, 76 insertions(+), 58 deletions(-) rename src/{active_observability_template => observability}/mod.rs (83%) rename src/{active_observability_template => observability}/pull.rs (99%) rename src/{active_observability_template => observability}/push.rs (99%) rename src/{active_observability_template => observability}/template.rs (100%) diff --git a/src/main.rs b/src/main.rs index 93fe9c16..e858f5f6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,7 +2,6 @@ use anyhow::{Context, Result}; use clap::{parser::ValueSource, ArgMatches, CommandFactory, FromArgMatches, Parser, Subcommand}; use std::ffi::{OsStr, OsString}; -mod active_observability_template; mod args; mod auth; #[allow(dead_code)] @@ -18,6 +17,7 @@ mod functions; mod http; mod init; mod js_runner; +mod observability; mod profiles; mod project_context; mod projects; @@ -61,36 +61,36 @@ const HELP_TEMPLATE: &str = "\ {before-help}{about} - {usage} Core - init Initialize .bt config directory and files - login Log in to Braintrust - logout Remove a saved Braintrust login - profiles Manage saved Braintrust login profiles - switch Switch org and project context - view View logs, traces, and spans + init Initialize .bt config directory and files + login Log in to Braintrust + logout Remove a saved Braintrust login + profiles Manage saved Braintrust login profiles + switch Switch org and project context + view View logs, traces, and spans Projects & resources - projects Manage projects - active-observability-template Pull and push portable active observability templates - topics Inspect and control Topics automation - prompts Manage prompts - functions Manage functions (tools, scorers, and more) - tools Manage tools - scorers Manage scorers - experiments Manage experiments - environments Manage deployment environments + projects Manage projects + observability Manage active observability tools + topics Inspect and control Topics automation + prompts Manage prompts + functions Manage functions (tools, scorers, and more) + tools Manage tools + scorers Manage scorers + experiments Manage experiments + environments Manage deployment environments Data & evaluation - datasets Manage datasets - eval Run eval files - sql Run SQL queries against Braintrust - sync Synchronize project logs between Braintrust and local NDJSON files + datasets Manage datasets + eval Run eval files + sql Run SQL queries against Braintrust + sync Synchronize project logs between Braintrust and local NDJSON files Additional - docs Manage workflow docs for coding agents - trace Manage coding-agent tracing - setup Configure Braintrust setup flows (deprecated: use curl -fsSL https://braintrust.dev/wizard/setup.sh | sh) - status Show current identity, org, and project context - update Update bt in-place + docs Manage workflow docs for coding agents + trace Manage coding-agent tracing + setup Configure Braintrust setup flows (deprecated: use curl -fsSL https://braintrust.dev/wizard/setup.sh | sh) + status Show current identity, org, and project context + update Update bt in-place Flags --profile Use a saved login profile [env: BRAINTRUST_PROFILE] @@ -151,10 +151,8 @@ enum Commands { Eval(CLIArgs), /// Manage projects Projects(CLIArgs), - /// Pull and push facets and Loop automations as a portable template - ActiveObservabilityTemplate( - CLIArgs, - ), + /// Manage active observability tools + Observability(CLIArgs), /// Inspect and control Topics automation Topics(CLIArgs), /// Manage datasets @@ -204,7 +202,7 @@ impl Commands { #[cfg(unix)] Commands::Eval(cmd) => &cmd.base, Commands::Projects(cmd) => &cmd.base, - Commands::ActiveObservabilityTemplate(cmd) => &cmd.base, + Commands::Observability(cmd) => &cmd.base, Commands::Topics(cmd) => &cmd.base, Commands::Datasets(cmd) => &cmd.base, Commands::Environments(cmd) => &cmd.base, @@ -236,7 +234,7 @@ impl Commands { #[cfg(unix)] Commands::Eval(cmd) => &mut cmd.base, Commands::Projects(cmd) => &mut cmd.base, - Commands::ActiveObservabilityTemplate(cmd) => &mut cmd.base, + Commands::Observability(cmd) => &mut cmd.base, Commands::Datasets(cmd) => &mut cmd.base, Commands::Environments(cmd) => &mut cmd.base, Commands::Topics(cmd) => &mut cmd.base, @@ -371,9 +369,7 @@ fn try_main() -> Result<()> { #[cfg(unix)] Commands::Eval(cmd) => eval::run(cmd.base, cmd.args).await?, Commands::Projects(cmd) => projects::run(cmd.base, cmd.args).await?, - Commands::ActiveObservabilityTemplate(cmd) => { - active_observability_template::run(cmd.base, cmd.args).await? - } + Commands::Observability(cmd) => observability::run(cmd.base, cmd.args).await?, Commands::Datasets(cmd) => datasets::run(cmd.base, cmd.args).await?, Commands::Environments(cmd) => environments::run(cmd.base, cmd.args).await?, Commands::Topics(cmd) => topics::run(cmd.base, cmd.args).await?, diff --git a/src/active_observability_template/mod.rs b/src/observability/mod.rs similarity index 83% rename from src/active_observability_template/mod.rs rename to src/observability/mod.rs index a9e6be38..0726260c 100644 --- a/src/active_observability_template/mod.rs +++ b/src/observability/mod.rs @@ -20,21 +20,33 @@ use crate::{ use self::{push::Snapshot, template::ActiveObservabilityTemplate}; +#[derive(Debug, Clone, Args)] +pub(crate) struct ObservabilityArgs { + #[command(subcommand)] + command: ObservabilityCommand, +} + +#[derive(Debug, Clone, Subcommand)] +enum ObservabilityCommand { + /// Pull and push facets and Loop automations as a portable template + Template(TemplateArgs), +} + #[derive(Debug, Clone, Args)] #[command(after_help = "\ Examples: - bt active-observability-template pull --output active-observability-template.json - bt active-observability-template push active-observability-template.json --project test-project - bt active-observability-template push https://example.com/active-observability-template.json - bt active-observability-template pull | bt active-observability-template push - --project test-project + bt observability template pull --output active-observability-template.json + bt observability template push active-observability-template.json --project test-project + bt observability template push https://example.com/active-observability-template.json + bt observability template pull | bt observability template push - --project test-project ")] -pub(crate) struct ActiveObservabilityTemplateArgs { +struct TemplateArgs { #[command(subcommand)] - command: ActiveObservabilityTemplateCommand, + command: TemplateCommand, } #[derive(Debug, Clone, Subcommand)] -enum ActiveObservabilityTemplateCommand { +enum TemplateCommand { /// Pull facets and Loop automations into a portable template Pull(PullArgs), /// Push facets and Loop automations from a portable template @@ -47,7 +59,7 @@ pub(super) struct PullArgs { #[arg( long, short = 'O', - env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PULL_OUTPUT", + env = "BT_OBSERVABILITY_TEMPLATE_PULL_OUTPUT", value_name = "PATH" )] output: Option, @@ -55,7 +67,7 @@ pub(super) struct PullArgs { /// Overwrite an existing output file #[arg( long, - env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PULL_FORCE", + env = "BT_OBSERVABILITY_TEMPLATE_PULL_FORCE", default_value_t = false, value_parser = clap::builder::BoolishValueParser::new() )] @@ -72,7 +84,7 @@ struct PushArgs { #[arg( long = "file", short = 'f', - env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_FILE", + env = "BT_OBSERVABILITY_TEMPLATE_PUSH_FILE", value_name = "SOURCE" )] source_flag: Option, @@ -80,7 +92,7 @@ struct PushArgs { /// Use this existing Topics automation for every facet #[arg( long, - env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_TOPICS_AUTOMATION", + env = "BT_OBSERVABILITY_TEMPLATE_PUSH_TOPICS_AUTOMATION", value_name = "NAME_OR_ID" )] topics_automation: Option, @@ -88,7 +100,7 @@ struct PushArgs { /// Replace existing matching resources #[arg( long, - env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_FORCE", + env = "BT_OBSERVABILITY_TEMPLATE_PUSH_FORCE", default_value_t = false, value_parser = clap::builder::BoolishValueParser::new() )] @@ -98,7 +110,7 @@ struct PushArgs { #[arg( long, short = 'y', - env = "BT_ACTIVE_OBSERVABILITY_TEMPLATE_PUSH_YES", + env = "BT_OBSERVABILITY_TEMPLATE_PUSH_YES", default_value_t = false, value_parser = clap::builder::BoolishValueParser::new() )] @@ -111,16 +123,22 @@ impl PushArgs { (Some(_), Some(_)) => bail!("use either a template source or --file, not both"), (Some(source), None) | (None, Some(source)) => Ok(source), (None, None) => bail!( - "active observability template source required. Use: bt active-observability-template push " + "active observability template source required. Use: bt observability template push " ), } } } -pub(crate) async fn run(base: BaseArgs, args: ActiveObservabilityTemplateArgs) -> Result<()> { +pub(crate) async fn run(base: BaseArgs, args: ObservabilityArgs) -> Result<()> { + match args.command { + ObservabilityCommand::Template(args) => run_template(base, args).await, + } +} + +async fn run_template(base: BaseArgs, args: TemplateArgs) -> Result<()> { match args.command { - ActiveObservabilityTemplateCommand::Pull(args) => pull::run(base, args).await, - ActiveObservabilityTemplateCommand::Push(args) => run_push(base, args).await, + TemplateCommand::Pull(args) => pull::run(base, args).await, + TemplateCommand::Push(args) => run_push(base, args).await, } } @@ -254,17 +272,19 @@ mod tests { #[test] fn active_observability_commands_parse_from_the_root_cli() { for args in [ - vec!["bt", "active-observability-template", "pull"], + vec!["bt", "observability", "template", "pull"], vec![ "bt", - "active-observability-template", + "observability", + "template", "pull", "--output", "template.json", ], vec![ "bt", - "active-observability-template", + "observability", + "template", "push", "template.json", "--topics-automation", @@ -274,13 +294,15 @@ mod tests { ], vec![ "bt", - "active-observability-template", + "observability", + "template", "push", "https://example.com/template.json", ], vec![ "bt", - "active-observability-template", + "observability", + "template", "push", "--file", "template.json", diff --git a/src/active_observability_template/pull.rs b/src/observability/pull.rs similarity index 99% rename from src/active_observability_template/pull.rs rename to src/observability/pull.rs index cd795872..76a30ec7 100644 --- a/src/active_observability_template/pull.rs +++ b/src/observability/pull.rs @@ -188,7 +188,7 @@ mod tests { use serde_json::json; use super::*; - use crate::active_observability_template::template::{KIND, SCHEMA_VERSION}; + use crate::observability::template::{KIND, SCHEMA_VERSION}; fn template() -> ActiveObservabilityTemplate { serde_json::from_value(json!({ diff --git a/src/active_observability_template/push.rs b/src/observability/push.rs similarity index 99% rename from src/active_observability_template/push.rs rename to src/observability/push.rs index dc0e4d13..508b213e 100644 --- a/src/active_observability_template/push.rs +++ b/src/observability/push.rs @@ -585,7 +585,7 @@ async fn upsert_function(client: &ApiClient, request: &Value, replace: bool) -> #[cfg(test)] mod tests { use super::*; - use crate::active_observability_template::template::{validate, KIND, SCHEMA_VERSION}; + use crate::observability::template::{validate, KIND, SCHEMA_VERSION}; fn facet(topics: Option<&str>) -> FacetTemplate { FacetTemplate { diff --git a/src/active_observability_template/template.rs b/src/observability/template.rs similarity index 100% rename from src/active_observability_template/template.rs rename to src/observability/template.rs From 754cfec35122f654984a6115c1240b6c1c787e4d Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 26 Aug 2026 16:46:02 -0700 Subject: [PATCH 4/4] test: tighten observability coverage --- src/observability/mod.rs | 8 -------- src/observability/pull.rs | 10 +--------- src/observability/push.rs | 8 +++++++- src/observability/template.rs | 5 ++++- 4 files changed, 12 insertions(+), 19 deletions(-) diff --git a/src/observability/mod.rs b/src/observability/mod.rs index 0726260c..16595ca0 100644 --- a/src/observability/mod.rs +++ b/src/observability/mod.rs @@ -272,7 +272,6 @@ mod tests { #[test] fn active_observability_commands_parse_from_the_root_cli() { for args in [ - vec!["bt", "observability", "template", "pull"], vec![ "bt", "observability", @@ -292,13 +291,6 @@ mod tests { "--force", "--yes", ], - vec![ - "bt", - "observability", - "template", - "push", - "https://example.com/template.json", - ], vec![ "bt", "observability", diff --git a/src/observability/pull.rs b/src/observability/pull.rs index 76a30ec7..2e5038a6 100644 --- a/src/observability/pull.rs +++ b/src/observability/pull.rs @@ -207,14 +207,6 @@ mod tests { .unwrap() } - #[test] - fn active_observability_stdout_is_only_pretty_json() { - let text = serialize_stdout(&template()).expect("serialize stdout"); - let value: serde_json::Value = serde_json::from_str(&text).expect("clean JSON"); - assert_eq!(value["kind"], KIND); - assert!(!text.contains("Pulled active observability")); - } - #[test] fn active_observability_selection_filters_both_resource_types() { let template = template(); @@ -224,7 +216,7 @@ mod tests { } #[test] - fn active_observability_noninteractive_pull_uses_active_defaults() { + fn active_observability_active_defaults_exclude_inactive_resources() { let mut template = template(); template.facets.push(FacetTemplate { name: "Active facet".to_string(), diff --git a/src/observability/push.rs b/src/observability/push.rs index 508b213e..710f9ff9 100644 --- a/src/observability/push.rs +++ b/src/observability/push.rs @@ -759,7 +759,7 @@ mod tests { } #[test] - fn active_observability_preflight_rejects_type_and_no_force_conflicts() { + fn active_observability_preflight_rejects_a_function_type_collision() { let wrong = plan( &template(facet(Some("Topics"))), Snapshot { @@ -771,7 +771,10 @@ mod tests { ) .expect_err("wrong type"); assert!(wrong.to_string().contains("not a facet")); + } + #[test] + fn active_observability_preflight_requires_force_for_an_existing_facet() { let no_force = plan( &template(facet(Some("Topics"))), Snapshot { @@ -783,7 +786,10 @@ mod tests { ) .expect_err("no force conflict"); assert!(no_force.to_string().contains("--force")); + } + #[test] + fn active_observability_preflight_rejects_an_automation_type_collision() { let wrong_loop = plan( &ActiveObservabilityTemplate { kind: KIND.to_string(), diff --git a/src/observability/template.rs b/src/observability/template.rs index 7319492c..10dc7195 100644 --- a/src/observability/template.rs +++ b/src/observability/template.rs @@ -814,7 +814,7 @@ mod tests { } #[test] - fn active_observability_force_conversions_preserve_customization_and_actions() { + fn active_observability_topic_map_reconciliation_preserves_customization() { let topic_map = function( "fn-test-topic-map", "test-facet-topic-map", @@ -842,7 +842,10 @@ mod tests { request["function_data"]["source_facet_function"], json!({"type": "function", "id": "fn-test-facet"}) ); + } + #[test] + fn active_observability_loop_replacement_preserves_target_actions() { let config = loop_config_for_target( &json!({"event_type": "windowed", "window": {}, "loop": {}, "actions": ["source"]}), Some(&json!({"event_type": "windowed", "loop": {}, "actions": ["target"]})),