From 9dea71c8d80d75df37660ed63028989984ca54fe Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Thu, 8 Oct 2026 15:22:56 -0700 Subject: [PATCH 1/7] fix: repair duplicate topic map wiring during template push --- README.md | 4 + src/observability/mod.rs | 31 ++++-- src/observability/push.rs | 177 ++++++++++++++++++++++++++++------ src/observability/template.rs | 6 ++ 4 files changed, 177 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index 12fb3372..daa24726 100644 --- a/README.md +++ b/README.md @@ -197,6 +197,10 @@ Remove-Item -Recurse -Force (Join-Path $userProfile ".braintrust\auth") -ErrorAc | `bt sync` | Synchronize project logs between Braintrust and local NDJSON files | | `bt update` | Update bt in-place | +## `bt observability template` + +Push with `--force` repairs duplicate topic map wiring by preserving the oldest map, breaking timestamp ties by function ID, and detaching extra maps from automations. The retained map keeps its customization and reports. Detached maps remain in the project and are listed in human and JSON push output. + ## `bt custom-views` Define custom views with `customTraceView` or `customDatasetView` from `braintrust/custom-views` (SDK 3.33.0 or later). diff --git a/src/observability/mod.rs b/src/observability/mod.rs index db45a662..3e95b359 100644 --- a/src/observability/mod.rs +++ b/src/observability/mod.rs @@ -182,16 +182,17 @@ async fn run_push(base: BaseArgs, args: PushArgs) -> Result<()> { .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, - }))? - ); + let mut output = serde_json::json!({ + "kind": "active_observability_template", + "status": "pushed", + "project": ctx.project.name, + "facets": result.facets, + "automations": result.automations, + }); + if !result.detached_topic_maps.is_empty() { + output["detached_topic_maps"] = serde_json::to_value(&result.detached_topic_maps)?; + } + println!("{}", serde_json::to_string(&output)?); } else { print_command_status( CommandStatus::Success, @@ -202,6 +203,16 @@ async fn run_push(base: BaseArgs, args: PushArgs) -> Result<()> { result.automations.len() ), ); + for map in &result.detached_topic_maps { + print_command_status( + CommandStatus::Warning, + &format!( + "Detached duplicate topic map '{}' ({}); it remains in the project", + map.slug.as_deref().unwrap_or(&map.name), + map.id + ), + ); + } } Ok(()) } diff --git a/src/observability/push.rs b/src/observability/push.rs index 740e44a8..a4c13b98 100644 --- a/src/observability/push.rs +++ b/src/observability/push.rs @@ -17,8 +17,8 @@ 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, remove_topics_functions, saved_preprocessor_slug, topic_map_ids, topic_map_matches, - topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, AutomationTemplate, - FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, + topic_map_order, topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, + AutomationTemplate, FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, }; #[derive(Debug)] @@ -34,6 +34,7 @@ pub(crate) struct MutationPlan { topics: BTreeMap, loops: Vec, function_ids: HashMap, + detached_topic_maps: Vec, } #[derive(Debug)] @@ -73,6 +74,8 @@ struct TopicsMutation { pub(crate) struct PushResult { pub facets: Vec, pub automations: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub detached_topic_maps: Vec, } #[derive(Debug, serde::Serialize)] @@ -132,6 +135,7 @@ pub(crate) fn plan( .filter(|automation| is_topics(&automation.config)) .collect::>(); let mut topics = BTreeMap::::new(); + let mut detached_topic_maps = BTreeMap::new(); let mut facets = Vec::with_capacity(template.facets.len()); for facet in &template.facets { @@ -150,12 +154,13 @@ pub(crate) fn plan( } let map_slug = topic_map_slug(&facet.slug); - let topic_map = existing_topic_map( + let (topic_map, duplicate_topic_maps) = existing_topic_map( existing, &topics_automations, &functions_by_id, &functions_by_slug, &map_slug, + force, )?; if let Some(topic_map) = topic_map { if topic_map @@ -199,12 +204,22 @@ pub(crate) fn plan( embedding_model: model, config, }); + for map in &duplicate_topic_maps { + detached_topic_maps + .entry(map.id.clone()) + .or_insert_with(|| PushedResource { + id: map.id.clone(), + name: map.name.clone(), + slug: Some(map.slug.clone()), + }); + } plan_topics_removals( &mut topics, &topics_automations, &key, existing, topic_map, + &duplicate_topic_maps, &functions_by_id, )?; facets.push(FacetMutation { @@ -254,6 +269,7 @@ pub(crate) fn plan( facets, topics, loops, + detached_topic_maps: detached_topic_maps.into_values().collect(), function_ids: snapshot .functions .into_iter() @@ -269,29 +285,45 @@ fn plan_topics_removals( selected_key: &str, facet: Option<&Function>, topic_map: Option<&Function>, + duplicate_topic_maps: &[&Function], functions_by_id: &HashMap<&str, &Function>, ) -> Result<()> { - if facet.is_none() && topic_map.is_none() { + if facet.is_none() && topic_map.is_none() && duplicate_topic_maps.is_empty() { return Ok(()); } for &automation in automations { let key = TopicsTarget::Existing(automation.clone()).key(); - if key == selected_key { - continue; - } - let current_config = topics + let selected = key == selected_key; + let mut 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 { + .map(|mutation| mutation.config.clone()) + .unwrap_or_else(|| automation.config.clone()); + let mut changed = false; + let facet_to_remove = (!selected) + .then(|| facet.map(|facet| facet.id.as_str())) + .flatten(); + let topic_map_to_remove = (!selected) + .then(|| topic_map.map(|topic_map| topic_map.id.as_str())) + .flatten(); + + if let Some(updated) = + remove_topics_functions(&config, facet_to_remove, topic_map_to_remove)? + { + config = updated; + changed = true; + } + for duplicate in duplicate_topic_maps { + if let Some(updated) = + remove_topics_functions(&config, None, Some(duplicate.id.as_str()))? + { + config = updated; + changed = true; + } + } + if !changed { continue; - }; + } match topics.get_mut(&key) { Some(mutation) => mutation.config = config, @@ -371,7 +403,8 @@ fn existing_topic_map<'a>( functions_by_id: &HashMap<&str, &'a Function>, functions_by_slug: &HashMap<&str, Vec<&'a Function>>, fallback_slug: &str, -) -> Result> { + force: bool, +) -> Result<(Option<&'a Function>, Vec<&'a Function>)> { if let Some(facet) = facet { let mut matches = BTreeMap::new(); for automation in topics_automations { @@ -392,22 +425,36 @@ fn existing_topic_map<'a>( } } } - match matches.len() { - 0 => {} - 1 => return Ok(matches.into_values().next()), - _ => bail!( - "facet '{}' is wired to multiple topic maps in the target project", - facet.slug - ), + if matches.len() == 1 { + return Ok((matches.into_values().next(), Vec::new())); + } + if matches.len() > 1 { + if !force { + bail!( + "facet '{}' is wired to multiple topic maps in the target project; use --force to preserve the oldest map and detach the duplicates", + facet.slug + ); + } + + let mut matches = matches.into_values().collect::>(); + // Older template-push versions could append a newly generated map without + // noticing an already-wired map with a different slug. Preserve the oldest + // map so existing reports and destination-owned customization survive. + matches.sort_by(|left, right| topic_map_order(left, right)); + let selected = matches.remove(0); + return Ok((Some(selected), matches)); } } - checked_function_by_slug( - functions_by_slug, - fallback_slug, - "classifier topic map", - "classifier", - ) + Ok(( + checked_function_by_slug( + functions_by_slug, + fallback_slug, + "classifier topic map", + "classifier", + )?, + Vec::new(), + )) } fn conflict(existing: Option<&Function>, label: &str, slug: &str, force: bool) -> Result<()> { @@ -630,6 +677,7 @@ pub(crate) async fn execute( Ok(PushResult { facets: pushed_facets, automations: pushed_loops, + detached_topic_maps: plan.detached_topic_maps, }) } @@ -890,6 +938,73 @@ mod tests { assert!(planned.topics.contains_key("id:auto-destination")); } + #[test] + fn active_observability_force_repairs_duplicate_topic_map_wiring() { + let mut existing_facet = existing_function("test-facet", "facet", "facet"); + existing_facet.id = "fn-existing-facet".to_string(); + + let mut original_topic_map = + existing_function("legacy-test-topic-map", "classifier", "topic_map"); + original_topic_map.id = "fn-original-topic-map".to_string(); + original_topic_map.created = Some("2026-01-01T00:00:00Z".to_string()); + original_topic_map.function_data = Some(json!({ + "type": "topic_map", + "source_facet_function": { + "type": "function", + "id": "fn-existing-facet" + }, + "generation_settings": {"max_topics": 12} + })); + + let mut duplicate_topic_map = + existing_function("test-facet-topic-map", "classifier", "topic_map"); + duplicate_topic_map.id = "fn-duplicate-topic-map".to_string(); + duplicate_topic_map.created = Some("2026-02-01T00:00:00Z".to_string()); + duplicate_topic_map.function_data = Some(json!({ + "type": "topic_map", + "source_facet_function": { + "type": "function", + "id": "fn-existing-facet" + } + })); + + let mut topics = automation("auto-topics", "Topics", "topic"); + topics.config["facet_functions"] = json!([{"type": "function", "id": "fn-existing-facet"}]); + topics.config["topic_map_functions"] = json!([ + {"function": {"type": "function", "id": "fn-original-topic-map"}}, + {"function": {"type": "function", "id": "fn-duplicate-topic-map"}} + ]); + + let planned = plan( + &template(facet(Some("Topics"))), + Snapshot { + functions: vec![existing_facet, original_topic_map, duplicate_topic_map], + automations: vec![topics], + }, + None, + true, + ) + .expect("force repairs duplicate topic map wiring"); + + let selected = planned.facets[0] + .topic_map + .as_ref() + .expect("oldest topic map is retained"); + assert_eq!(selected.id, "fn-original-topic-map"); + assert_eq!(planned.detached_topic_maps.len(), 1); + assert_eq!(planned.detached_topic_maps[0].id, "fn-duplicate-topic-map"); + assert_eq!( + selected.function_data.as_ref().unwrap()["generation_settings"], + json!({"max_topics": 12}) + ); + assert_eq!( + planned.topics["id:auto-topics"].config["topic_map_functions"], + json!([ + {"function": {"type": "function", "id": "fn-original-topic-map"}} + ]) + ); + } + #[test] fn active_observability_preflight_rejects_a_function_type_collision() { let wrong = plan( diff --git a/src/observability/template.rs b/src/observability/template.rs index c15e6ca7..ee186252 100644 --- a/src/observability/template.rs +++ b/src/observability/template.rs @@ -232,6 +232,12 @@ fn topics_by_facet( .collect() } +pub(crate) fn topic_map_order(left: &Function, right: &Function) -> std::cmp::Ordering { + left.created + .cmp(&right.created) + .then_with(|| left.id.cmp(&right.id)) +} + pub(crate) fn topic_map_matches(topic_map: &Function, facet: &Function) -> bool { let Some(data) = topic_map.function_data.as_ref() else { return false; From a027f8dd55f0558ef0f70768478147069a5d366c Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Tue, 13 Oct 2026 13:10:00 -0700 Subject: [PATCH 2/7] fix: replace topic map bindings only in the destination automation --- README.md | 2 +- src/observability/push.rs | 151 +++++++++++----------------------- src/observability/template.rs | 6 -- 3 files changed, 49 insertions(+), 110 deletions(-) diff --git a/README.md b/README.md index 97ccf1f8..5ed747aa 100644 --- a/README.md +++ b/README.md @@ -214,7 +214,7 @@ Remove-Item -Recurse -Force (Join-Path $userProfile ".braintrust\auth") -ErrorAc ## `bt observability template` -Push with `--force` repairs duplicate topic map wiring by preserving the oldest map, breaking timestamp ties by function ID, and detaching extra maps from automations. The retained map keeps its customization and reports. Detached maps remain in the project and are listed in human and JSON push output. +Push with `--force` replaces an imported facet's topic map binding in the selected automation. It reuses the map with the generated template slug, or creates that map if needed, and detaches superseded maps from that automation. Other automations keep their facet and map bindings. Detached map functions remain in the project and are listed in human and JSON push output. Existing generated state updates when the Topics pipeline runs again. ## `bt custom-views` diff --git a/src/observability/push.rs b/src/observability/push.rs index a4c13b98..0ac1c54e 100644 --- a/src/observability/push.rs +++ b/src/observability/push.rs @@ -17,8 +17,8 @@ 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, remove_topics_functions, saved_preprocessor_slug, topic_map_ids, topic_map_matches, - topic_map_order, topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, - AutomationTemplate, FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, + topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, AutomationTemplate, + FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, }; #[derive(Debug)] @@ -153,10 +153,21 @@ pub(crate) fn plan( } } + let target = resolve_topics_target( + facet, + topics_override, + &snapshot.automations, + &topics_automations, + &automations_by_name, + )?; + let binding_automations = match &target { + TopicsTarget::Existing(automation) => vec![automation], + TopicsTarget::New(_) => Vec::new(), + }; let map_slug = topic_map_slug(&facet.slug); let (topic_map, duplicate_topic_maps) = existing_topic_map( existing, - &topics_automations, + &binding_automations, &functions_by_id, &functions_by_slug, &map_slug, @@ -182,13 +193,6 @@ pub(crate) fn plan( 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), @@ -281,62 +285,20 @@ pub(crate) fn plan( fn plan_topics_removals( topics: &mut BTreeMap, - automations: &[&ProjectAutomation], + _automations: &[&ProjectAutomation], selected_key: &str, - facet: Option<&Function>, - topic_map: Option<&Function>, + _facet: Option<&Function>, + _topic_map: Option<&Function>, duplicate_topic_maps: &[&Function], - functions_by_id: &HashMap<&str, &Function>, + _functions_by_id: &HashMap<&str, &Function>, ) -> Result<()> { - if facet.is_none() && topic_map.is_none() && duplicate_topic_maps.is_empty() { - return Ok(()); - } - - for &automation in automations { - let key = TopicsTarget::Existing(automation.clone()).key(); - let selected = key == selected_key; - let mut config = topics - .get(&key) - .map(|mutation| mutation.config.clone()) - .unwrap_or_else(|| automation.config.clone()); - let mut changed = false; - let facet_to_remove = (!selected) - .then(|| facet.map(|facet| facet.id.as_str())) - .flatten(); - let topic_map_to_remove = (!selected) - .then(|| topic_map.map(|topic_map| topic_map.id.as_str())) - .flatten(); - - if let Some(updated) = - remove_topics_functions(&config, facet_to_remove, topic_map_to_remove)? + let selected = topics + .get_mut(selected_key) + .expect("resolved Topics destination"); + for duplicate in duplicate_topic_maps { + if let Some(config) = remove_topics_functions(&selected.config, None, Some(&duplicate.id))? { - config = updated; - changed = true; - } - for duplicate in duplicate_topic_maps { - if let Some(updated) = - remove_topics_functions(&config, None, Some(duplicate.id.as_str()))? - { - config = updated; - changed = true; - } - } - if !changed { - 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, - }, - ); - } + selected.config = config; } } Ok(()) @@ -425,25 +387,25 @@ fn existing_topic_map<'a>( } } } - if matches.len() == 1 { - return Ok((matches.into_values().next(), Vec::new())); - } - if matches.len() > 1 { - if !force { - bail!( - "facet '{}' is wired to multiple topic maps in the target project; use --force to preserve the oldest map and detach the duplicates", - facet.slug - ); + let selected = checked_function_by_slug( + functions_by_slug, + fallback_slug, + "classifier topic map", + "classifier", + )?; + if let Some(selected) = selected { + if !topic_map_matches(selected, facet) { + bail!("topic map slug '{fallback_slug}' belongs to a different facet; use a different facet slug"); } - - let mut matches = matches.into_values().collect::>(); - // Older template-push versions could append a newly generated map without - // noticing an already-wired map with a different slug. Preserve the oldest - // map so existing reports and destination-owned customization survive. - matches.sort_by(|left, right| topic_map_order(left, right)); - let selected = matches.remove(0); - return Ok((Some(selected), matches)); } + let duplicates = matches + .into_values() + .filter(|map| selected.is_none_or(|selected| map.id != selected.id)) + .collect::>(); + if !duplicates.is_empty() && !force { + bail!("facet '{}' has existing topic map bindings; use --force to replace the selected automation's binding", facet.slug); + } + return Ok((selected, duplicates)); } Ok(( @@ -883,7 +845,7 @@ mod tests { } #[test] - fn active_observability_finds_and_remaps_an_existing_topic_map() { + fn active_observability_preserves_other_automation_bindings() { let mut existing_facet = existing_function("test-facet", "facet", "facet"); existing_facet.id = "fn-existing-facet".to_string(); let mut existing_topic_map = existing_function("test-facet", "classifier", "topic_map"); @@ -920,21 +882,8 @@ mod tests { ) .expect("existing topic map is found through Topics wiring"); - let topic_map = planned.facets[0] - .topic_map - .as_ref() - .expect("existing topic map"); - assert_eq!(topic_map.id, "fn-existing-topic-map"); - assert_eq!(topic_map.slug, "test-facet"); - 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.facets[0].topic_map.is_none()); + assert!(!planned.topics.contains_key("id:auto-previous")); assert!(planned.topics.contains_key("id:auto-destination")); } @@ -989,18 +938,14 @@ mod tests { let selected = planned.facets[0] .topic_map .as_ref() - .expect("oldest topic map is retained"); - assert_eq!(selected.id, "fn-original-topic-map"); + .expect("template map is retained"); + assert_eq!(selected.id, "fn-duplicate-topic-map"); assert_eq!(planned.detached_topic_maps.len(), 1); - assert_eq!(planned.detached_topic_maps[0].id, "fn-duplicate-topic-map"); - assert_eq!( - selected.function_data.as_ref().unwrap()["generation_settings"], - json!({"max_topics": 12}) - ); + assert_eq!(planned.detached_topic_maps[0].id, "fn-original-topic-map"); assert_eq!( planned.topics["id:auto-topics"].config["topic_map_functions"], json!([ - {"function": {"type": "function", "id": "fn-original-topic-map"}} + {"function": {"type": "function", "id": "fn-duplicate-topic-map"}} ]) ); } diff --git a/src/observability/template.rs b/src/observability/template.rs index ee186252..c15e6ca7 100644 --- a/src/observability/template.rs +++ b/src/observability/template.rs @@ -232,12 +232,6 @@ fn topics_by_facet( .collect() } -pub(crate) fn topic_map_order(left: &Function, right: &Function) -> std::cmp::Ordering { - left.created - .cmp(&right.created) - .then_with(|| left.id.cmp(&right.id)) -} - pub(crate) fn topic_map_matches(topic_map: &Function, facet: &Function) -> bool { let Some(data) = topic_map.function_data.as_ref() else { return false; From e37529151e3bdda7b123351cc8bb69ce6bc84d3b Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 14 Oct 2026 10:05:41 -0700 Subject: [PATCH 3/7] feat: carry Topics automation settings in observability templates --- README.md | 15 ++ src/observability/mod.rs | 6 +- src/observability/pull.rs | 7 +- src/observability/push.rs | 315 ++++++++++++++++++++++++++++++++-- src/observability/template.rs | 161 ++++++++++++++++- 5 files changed, 486 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index a0f3aaa3..da28b160 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,21 @@ Remove-Item -Recurse -Force (Join-Path $userProfile ".braintrust\auth") -ErrorAc | `bt sync` | Synchronize project logs between Braintrust and local NDJSON files | | `bt update` | Update bt in-place | +## `bt observability template` + +Copy custom facets, their Topics automation settings, and Loop automations between projects: + +```bash +bt observability template pull --output observability-template.json --project test-source-project +bt observability template push observability-template.json --project test-destination-project +``` + +Schema v2 adds `topics_automations`, with each referenced automation's name, description, and API-returned config. Scope, automation-level BTQL filters, sampling, model, status, and timing travel with the facet. Project-specific `facet_functions` and `topic_map_functions` wiring is rebuilt in the destination. Version 1 templates remain readable and use destination/default Topics settings. + +New automations use the bundled settings. Existing automations with matching settings and descriptions retain them. Differences require `--force`, which replaces settings in full while preserving destination facet/map wiring; destination-only settings are dropped. `--topics-automation ` uses that existing automation's settings instead of the template's. Existing resource replacement rules still apply to facet and preprocessor definitions. + +Settings scoped to a specific experiment cannot be exported because its ID is project-specific. Timing validation and normalization remain with the server; settings returned by the API may differ from the submitted values. Topic map definitions and generated state are not exported. Topics update when the pipeline runs again. + ## `bt custom-views` Define custom views with `customTraceView` or `customDatasetView` from `braintrust/custom-views` (SDK 3.33.0 or later). diff --git a/src/observability/mod.rs b/src/observability/mod.rs index a808c595..3d67ce89 100644 --- a/src/observability/mod.rs +++ b/src/observability/mod.rs @@ -28,7 +28,7 @@ pub(crate) struct ObservabilityArgs { #[derive(Debug, Clone, Subcommand)] enum ObservabilityCommand { - /// Pull and push facets and Loop automations as a portable template + /// Pull and push facets, Topics settings, and Loop automations as a portable template Template(TemplateArgs), } @@ -47,9 +47,9 @@ struct TemplateArgs { #[derive(Debug, Clone, Subcommand)] enum TemplateCommand { - /// Pull facets and Loop automations into a portable template + /// Pull facets, Topics settings, and Loop automations into a portable template Pull(PullArgs), - /// Push facets and Loop automations from a portable template + /// Push facets, Topics settings, and Loop automations from a portable template Push(PushArgs), } diff --git a/src/observability/pull.rs b/src/observability/pull.rs index 9e7c7e0a..1058ea8a 100644 --- a/src/observability/pull.rs +++ b/src/observability/pull.rs @@ -15,8 +15,8 @@ use crate::{ use super::{ template::{ - deduplicate_preprocessors, from_remote, ActiveObservabilityTemplate, AutomationTemplate, - FacetTemplate, + deduplicate_preprocessors, from_remote, retain_topics_settings, validate, + validate_topics_scope, ActiveObservabilityTemplate, AutomationTemplate, FacetTemplate, }, PullArgs, }; @@ -41,6 +41,9 @@ pub(crate) async fn run(base: BaseArgs, args: PullArgs) -> Result<()> { (template.facets, template.automations) = filter_active_resources(template.facets, template.automations); } + retain_topics_settings(&mut template); + validate(&template)?; + validate_topics_scope(&template)?; // Selection happens first so a selected facet never loses its required definition. deduplicate_preprocessors(&mut template.facets); diff --git a/src/observability/push.rs b/src/observability/push.rs index 740e44a8..609c78fd 100644 --- a/src/observability/push.rs +++ b/src/observability/push.rs @@ -17,8 +17,9 @@ 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, remove_topics_functions, saved_preprocessor_slug, topic_map_ids, topic_map_matches, - topic_map_slug, with_preprocessor_id, ActiveObservabilityTemplate, AutomationTemplate, - FacetTemplate, PortableFunction, DEFAULT_TOPICS_DESCRIPTION, + topic_map_slug, topics_config_for_target, topics_settings, validate_topics_scope, + with_preprocessor_id, ActiveObservabilityTemplate, AutomationTemplate, FacetTemplate, + PortableFunction, DEFAULT_TOPICS_DESCRIPTION, }; #[derive(Debug)] @@ -67,6 +68,7 @@ struct TopicsMutation { target: TopicsTarget, embedding_model: String, config: Value, + description: Option, } #[derive(Debug, Serialize)] @@ -89,6 +91,9 @@ pub(crate) fn plan( topics_override: Option<&str>, force: bool, ) -> Result { + if topics_override.is_none() { + validate_topics_scope(template)?; + } let functions_by_slug = functions_by_slug(&snapshot.functions); let functions_by_id = snapshot .functions @@ -194,10 +199,15 @@ pub(crate) fn plan( TopicsTarget::Existing(automation) => add_topics_functions(&automation.config, &[])?, TopicsTarget::New(_) => default_topics_config(), }; + let description = match &target { + TopicsTarget::Existing(automation) => automation.description.clone(), + TopicsTarget::New(_) => Some(DEFAULT_TOPICS_DESCRIPTION.to_string()), + }; topics.entry(key.clone()).or_insert(TopicsMutation { target, embedding_model: model, config, + description, }); plan_topics_removals( &mut topics, @@ -215,6 +225,32 @@ pub(crate) fn plan( }); } + if topics_override.is_none() { + for mutation in topics.values_mut() { + let name = match &mutation.target { + TopicsTarget::Existing(automation) => &automation.name, + TopicsTarget::New(name) => name, + }; + let Some(settings) = template + .topics_automations + .iter() + .find(|settings| &settings.name == name) + else { + continue; + }; + if let TopicsTarget::Existing(existing) = &mutation.target { + if !force + && (topics_settings(&existing.config)? != settings.config + || existing.description != settings.description) + { + bail!("Topics automation '{name}' has different settings; use --force to replace them or --topics-automation to use the destination settings"); + } + } + mutation.config = topics_config_for_target(&settings.config, &mutation.config)?; + mutation.description = settings.description.clone(); + } + } + let loops = template .automations .iter() @@ -299,6 +335,7 @@ fn plan_topics_removals( topics.insert( key, TopicsMutation { + description: automation.description.clone(), target: TopicsTarget::Existing(automation.clone()), embedding_model: embedding_model(automation, functions_by_id), config, @@ -559,7 +596,7 @@ pub(crate) async fn execute( TopicsTarget::Existing(automation) => { let body = json!({ "name": automation.name, - "description": automation.description, + "description": mutation.description, "config": config, }); patch_project_automation(client, &automation.id, &body) @@ -572,7 +609,7 @@ pub(crate) async fn execute( let body = json!({ "project_id": project_id, "name": name, - "description": DEFAULT_TOPICS_DESCRIPTION, + "description": mutation.description, "config": config, }); let created = create_project_automation(client, &body) @@ -644,9 +681,7 @@ async fn upsert_function(client: &ApiClient, request: &Value, replace: bool) -> #[cfg(test)] mod tests { use super::*; - use crate::observability::template::{ - deduplicate_preprocessors, validate, KIND, SCHEMA_VERSION, - }; + use crate::observability::template::{deduplicate_preprocessors, validate, KIND}; fn facet(topics: Option<&str>) -> FacetTemplate { FacetTemplate { @@ -665,8 +700,9 @@ mod tests { fn template(facet: FacetTemplate) -> ActiveObservabilityTemplate { ActiveObservabilityTemplate { kind: KIND.to_string(), - schema_version: SCHEMA_VERSION, + schema_version: 1, facets: vec![facet], + topics_automations: Vec::new(), automations: Vec::new(), } } @@ -813,8 +849,9 @@ mod tests { deduplicate_preprocessors(&mut facets); let source = ActiveObservabilityTemplate { kind: KIND.to_string(), - schema_version: SCHEMA_VERSION, + schema_version: 1, facets, + topics_automations: Vec::new(), automations: Vec::new(), }; validate(&source).expect("valid shared preprocessor template"); @@ -925,8 +962,9 @@ mod tests { let wrong_loop = plan( &ActiveObservabilityTemplate { kind: KIND.to_string(), - schema_version: SCHEMA_VERSION, + schema_version: 1, facets: Vec::new(), + topics_automations: Vec::new(), automations: vec![AutomationTemplate { name: "Test automation".to_string(), description: None, @@ -943,4 +981,261 @@ mod tests { .expect_err("wrong automation type"); assert!(wrong_loop.to_string().contains("not a Loop")); } + + fn scoped_template() -> ActiveObservabilityTemplate { + let mut source = template(facet(Some("Test scoped Topics"))); + source.schema_version = 2; + source.topics_automations = vec![AutomationTemplate { + name: "Test scoped Topics".to_string(), + description: Some("Test span policy".to_string()), + config: json!({"event_type":"topic", "scope":{"type":"span"}, + "data_scope":{"type":"project_logs"}, "btql_filter":"metadata.test_enabled = true", + "sampling_rate":1, "model":"test-model", "status":"active", "rerun_seconds":86400, + "backfill_time_range":"604800s"}), + }]; + source + } + + #[test] + fn active_observability_v2_pull_push_round_trip_creates_bundled_settings() { + use crate::observability::template::from_remote; + let expected = scoped_template(); + let settings = &expected.topics_automations[0]; + let source_facet = existing_function("test-facet", "facet", "facet"); + let mut source_map = existing_function("test-facet-topic-map", "classifier", "topic_map"); + source_map.function_data = Some( + json!({"type":"topic_map", "source_facet_function":{"type":"function", "id":source_facet.id}}), + ); + let mut remote = automation("test-source-automation", &settings.name, "topic"); + remote.description = settings.description.clone(); + remote.config = add_topics_functions( + &topics_config_for_target(&settings.config, &json!({})).unwrap(), + &[(source_facet.id.clone(), source_map.id.clone())], + ) + .unwrap(); + let pulled = from_remote(&[source_facet, source_map], &[remote]).unwrap(); + let source = serde_json::from_str(&serde_json::to_string(&pulled).unwrap()).unwrap(); + validate(&source).unwrap(); + assert_eq!(source.topics_automations, expected.topics_automations); + let planned = plan( + &source, + Snapshot { + functions: vec![], + automations: vec![], + }, + None, + false, + ) + .unwrap(); + let mutation = &planned.topics["new:Test scoped Topics"]; + assert_eq!(topics_settings(&mutation.config).unwrap(), settings.config); + assert_eq!(mutation.description, settings.description); + assert_eq!(mutation.config["facet_functions"], json!([])); + assert_eq!(mutation.config["topic_map_functions"], json!([])); + } + + #[test] + fn active_observability_v2_matching_settings_do_not_require_force() { + let source = scoped_template(); + let settings = &source.topics_automations[0]; + let mut existing = automation("test-existing-automation", &settings.name, "topic"); + existing.config = topics_config_for_target(&settings.config, &json!({})).unwrap(); + existing.description = settings.description.clone(); + let planned = plan( + &source, + Snapshot { + functions: vec![], + automations: vec![existing.clone()], + }, + None, + false, + ) + .unwrap(); + assert_eq!( + planned.topics["id:test-existing-automation"].config, + existing.config + ); + assert_eq!( + planned.topics["id:test-existing-automation"].description, + existing.description + ); + } + + #[test] + fn active_observability_v2_differing_settings_require_force_and_keep_wiring() { + let mut source = scoped_template(); + source.topics_automations[0].description = None; + let mut existing = automation("test-existing-automation", "Test scoped Topics", "topic"); + existing.description = Some("Test destination description".to_string()); + existing.config = add_topics_functions( + &default_topics_config(), + &[( + "fn-test-other-facet".to_string(), + "fn-test-other-map".to_string(), + )], + ) + .unwrap(); + existing.config["test_destination_only"] = json!(true); + let snapshot = || Snapshot { + functions: vec![], + automations: vec![existing.clone()], + }; + assert!(plan(&source, snapshot(), None, false) + .unwrap_err() + .to_string() + .contains("--force")); + let planned = plan(&source, snapshot(), None, true).unwrap(); + let mutation = &planned.topics["id:test-existing-automation"]; + assert_eq!( + topics_settings(&mutation.config).unwrap(), + source.topics_automations[0].config + ); + assert_eq!( + mutation.config["facet_functions"], + existing.config["facet_functions"] + ); + assert_eq!( + mutation.config["topic_map_functions"], + existing.config["topic_map_functions"] + ); + assert!(mutation.description.is_none()); + } + + #[test] + fn active_observability_v2_override_uses_destination_settings() { + let mut source = scoped_template(); + source.topics_automations[0].config["data_scope"] = + json!({"type":"experiment", "experiment_id":"test-experiment-id"}); + source.topics_automations[0].config["backfill_time_range"] = json!("server-owned-value"); + let existing = automation( + "test-destination-automation", + "Test destination Topics", + "topic", + ); + let planned = plan( + &source, + Snapshot { + functions: vec![], + automations: vec![existing.clone()], + }, + Some("test-destination-automation"), + false, + ) + .unwrap(); + assert_eq!(planned.topics.len(), 1); + let mutation = &planned.topics["id:test-destination-automation"]; + assert_eq!(mutation.config, existing.config); + assert_eq!(mutation.description, existing.description); + } + + #[tokio::test] + async fn active_observability_v2_execute_sends_settings_description_and_wiring() { + use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer}; + use std::{ + net::TcpListener, + sync::{Arc, Mutex}, + }; + + type RecordedRequests = Arc>>; + let requests: RecordedRequests = Arc::new(Mutex::new(Vec::new())); + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let api_url = format!("http://{}", listener.local_addr().unwrap()); + let state = web::Data::new(requests.clone()); + let server = HttpServer::new(move || { + App::new().app_data(state.clone()).default_service(web::to( + |request: HttpRequest, + body: web::Json, + state: web::Data| async move { + let mut response = body.into_inner(); + state.lock().unwrap().push(( + request.method().to_string(), + request.path().to_string(), + response.clone(), + )); + match (request.method().as_str(), request.path()) { + ("PUT", "/v1/function") => { + response["id"] = if response["function_type"] == "facet" { + json!("fn-test-facet") + } else { + json!("fn-test-map") + }; + } + ("PATCH", "/v1/project_automation/test-existing-automation") => { + response["id"] = json!("test-existing-automation"); + response["project_id"] = json!("test-project-id"); + } + _ => return HttpResponse::NotFound().finish(), + } + HttpResponse::Ok().json(response) + }, + )) + }) + .workers(1) + .listen(listener) + .unwrap() + .run(); + let handle = server.handle(); + tokio::spawn(server); + let login = braintrust_sdk_rust::LoginState::new(); + login.set( + "test-key".to_string(), + "test-org-id".to_string(), + "test-org".to_string(), + api_url.clone(), + "https://example.invalid".to_string(), + ); + let client = ApiClient::new(&crate::auth::LoginContext { + login, + api_url, + app_url: "https://example.invalid".to_string(), + profile: None, + }) + .unwrap(); + + let mut source = scoped_template(); + source.topics_automations[0].description = None; + let mut facet = existing_function("test-facet", "facet", "facet"); + facet.id = "fn-test-facet".to_string(); + let mut map = existing_function("test-facet-topic-map", "classifier", "topic_map"); + map.id = "fn-test-map".to_string(); + map.function_data = Some( + json!({"type": "topic_map", "source_facet_function": {"type": "function", "id": facet.id}}), + ); + let mut automation = automation("test-existing-automation", "Test scoped Topics", "topic"); + automation.description = Some("Test old description".to_string()); + automation.config = + add_topics_functions(&automation.config, &[(facet.id.clone(), map.id.clone())]) + .unwrap(); + let planned = plan( + &source, + Snapshot { + functions: vec![facet, map], + automations: vec![automation], + }, + None, + true, + ) + .unwrap(); + let result = execute(&client, "test-project-id", planned).await; + handle.stop(true).await; + assert_eq!(result.unwrap().facets[0].id, "fn-test-facet"); + let requests = requests.lock().unwrap(); + assert_eq!(requests.len(), 3); + assert_eq!(requests[0].0, "PUT"); + assert_eq!(requests[1].0, "PUT"); + assert_eq!(requests[2].0, "PATCH"); + let body = &requests[2].2; + assert_eq!(body["name"], "Test scoped Topics"); + assert!(body["description"].is_null()); + assert_eq!(body["config"]["scope"], json!({"type": "span"})); + assert_eq!( + body["config"]["btql_filter"], + "metadata.test_enabled = true" + ); + assert_eq!(body["config"]["facet_functions"][0]["id"], "fn-test-facet"); + assert_eq!( + body["config"]["topic_map_functions"][0]["function"]["id"], + "fn-test-map" + ); + } } diff --git a/src/observability/template.rs b/src/observability/template.rs index c15e6ca7..5ad30bed 100644 --- a/src/observability/template.rs +++ b/src/observability/template.rs @@ -7,10 +7,11 @@ 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 SCHEMA_VERSION: u32 = 2; 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"; +const WIRING_KEYS: [&str; 2] = ["facet_functions", "topic_map_functions"]; #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] pub(crate) struct ActiveObservabilityTemplate { @@ -18,6 +19,8 @@ pub(crate) struct ActiveObservabilityTemplate { pub schema_version: u32, #[serde(default)] pub facets: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub topics_automations: Vec, #[serde(default)] pub automations: Vec, } @@ -97,6 +100,25 @@ pub(crate) fn from_remote( .collect::>>()?; facet_templates.sort_by(|a, b| a.name.cmp(&b.name).then(a.slug.cmp(&b.slug))); + let names = facet_templates + .iter() + .filter_map(|facet| facet.topics_automation.as_deref()) + .collect::>(); + let mut topics_templates = automations + .iter() + .filter(|automation| { + is_topics(&automation.config) && names.contains(automation.name.as_str()) + }) + .map(|automation| { + Ok(AutomationTemplate { + name: automation.name.clone(), + description: automation.description.clone(), + config: topics_settings(&automation.config)?, + }) + }) + .collect::>>()?; + topics_templates.sort_by(|a, b| a.name.cmp(&b.name)); + let mut loop_templates = automations .iter() .filter(|automation| is_loop_config(&automation.config)) @@ -116,6 +138,7 @@ pub(crate) fn from_remote( kind: KIND.to_string(), schema_version: SCHEMA_VERSION, facets: facet_templates, + topics_automations: topics_templates, automations: loop_templates, }) } @@ -255,9 +278,9 @@ pub(crate) fn validate(template: &ActiveObservabilityTemplate) -> Result<()> { if template.kind != KIND { bail!("template kind must be '{KIND}'"); } - if template.schema_version != SCHEMA_VERSION { + if !matches!(template.schema_version, 1 | SCHEMA_VERSION) { bail!( - "unsupported template schema version {}; supported version is {SCHEMA_VERSION}", + "unsupported template schema version {}; supported versions are 1 and {SCHEMA_VERSION}", template.schema_version ); } @@ -311,7 +334,40 @@ pub(crate) fn validate(template: &ActiveObservabilityTemplate) -> Result<()> { } } + if template.schema_version == 1 && !template.topics_automations.is_empty() { + bail!("bundled Topics settings require template schema version 2"); + } let mut automation_names = HashSet::new(); + for automation in &template.topics_automations { + require_text(&automation.name, "Topics automation name")?; + if !automation_names.insert(automation.name.as_str()) { + bail!( + "template contains duplicate automation name '{}'", + automation.name + ); + } + if !is_topics(&automation.config) { + bail!( + "automation '{}' is not a Topics automation (expected event_type 'topic')", + automation.name + ); + } + if WIRING_KEYS + .iter() + .any(|key| automation.config.get(key).is_some()) + { + bail!("Topics automation '{}' settings must not contain facet_functions or topic_map_functions", automation.name); + } + } + if template.schema_version == SCHEMA_VERSION { + for facet in &template.facets { + if let Some(name) = facet.topics_automation.as_deref() { + if !automation_names.contains(name) { + bail!("facet '{}' references Topics automation '{name}' without bundled settings in topics_automations", facet.name); + } + } + } + } for automation in &template.automations { require_text(&automation.name, "automation name")?; if !automation_names.insert(automation.name.as_str()) { @@ -330,6 +386,56 @@ pub(crate) fn validate(template: &ActiveObservabilityTemplate) -> Result<()> { Ok(()) } +pub(crate) fn validate_topics_scope(template: &ActiveObservabilityTemplate) -> Result<()> { + for automation in &template.topics_automations { + if automation + .config + .get("data_scope") + .and_then(|scope| scope.get("experiment_id")) + .is_some_and(|id| !id.is_null()) + { + let facet = template + .facets + .iter() + .find(|facet| facet.topics_automation.as_deref() == Some(automation.name.as_str())); + if let Some(facet) = facet { + bail!("facet '{}' uses Topics automation '{}' scoped to a specific experiment; its settings cannot be copied to another project", facet.name, automation.name); + } + } + } + Ok(()) +} + +pub(crate) fn retain_topics_settings(template: &mut ActiveObservabilityTemplate) { + let names = template + .facets + .iter() + .filter_map(|facet| facet.topics_automation.as_deref()) + .collect::>(); + template + .topics_automations + .retain(|automation| names.contains(automation.name.as_str())); +} + +pub(crate) fn topics_settings(config: &Value) -> Result { + let mut settings = object(config, "Topics automation config")?.clone(); + for key in WIRING_KEYS { + settings.remove(key); + } + Ok(Value::Object(settings)) +} + +pub(crate) fn topics_config_for_target(settings: &Value, destination: &Value) -> Result { + let mut config = object(settings, "Topics automation settings")?.clone(); + for key in WIRING_KEYS { + config.insert( + key.to_string(), + destination.get(key).cloned().unwrap_or_else(|| json!([])), + ); + } + Ok(Value::Object(config)) +} + fn reserve_slug( slugs: &mut HashMap, slug: &str, @@ -818,4 +924,53 @@ mod tests { assert_eq!(twice["facet_functions"].as_array().unwrap().len(), 1); assert_eq!(twice["topic_map_functions"].as_array().unwrap().len(), 1); } + + #[test] + fn active_observability_v2_settings_validate_and_follow_selection() { + let mut source: ActiveObservabilityTemplate = serde_json::from_value(json!({ + "kind": KIND, "schema_version": 2, + "facets": [{"name":"Test facet", "slug":"test-facet", "topics_automation":"Test Topics", "function_data":{"type":"facet"}}], + "topics_automations": [{"name":"Test Topics", "config":{"event_type":"topic", "backfill_time_range":"server-owned-value"}}] + })).unwrap(); + validate(&source).unwrap(); + source.topics_automations[0].config["facet_functions"] = json!([]); + assert!(validate(&source) + .unwrap_err() + .to_string() + .contains("facet_functions")); + source.topics_automations[0] + .config + .as_object_mut() + .unwrap() + .remove("facet_functions"); + source.topics_automations[0].config["event_type"] = json!("windowed"); + assert!(validate(&source) + .unwrap_err() + .to_string() + .contains("not a Topics")); + source.topics_automations[0].config["event_type"] = json!("topic"); + source + .topics_automations + .push(source.topics_automations[0].clone()); + assert!(validate(&source) + .unwrap_err() + .to_string() + .contains("duplicate")); + source.topics_automations.pop(); + let mut missing = source.clone(); + missing.topics_automations.clear(); + assert!(validate(&missing) + .unwrap_err() + .to_string() + .contains("without bundled settings")); + missing.schema_version = 1; + validate(&missing).unwrap(); + source.topics_automations[0].config["data_scope"] = + json!({"type":"experiment", "experiment_id":"test-experiment-id"}); + let error = validate_topics_scope(&source).unwrap_err().to_string(); + assert!(error.contains("Test facet") && error.contains("specific experiment")); + source.facets.clear(); + retain_topics_settings(&mut source); + assert!(source.topics_automations.is_empty()); + } } From 70a9fe629c723d03ed098eb5feb39e5eb0299cc6 Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 14 Oct 2026 10:30:27 -0700 Subject: [PATCH 4/7] refactor: remove unused Topics removal planner arguments --- src/observability/push.rs | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/src/observability/push.rs b/src/observability/push.rs index 0ac1c54e..a479bf6e 100644 --- a/src/observability/push.rs +++ b/src/observability/push.rs @@ -217,15 +217,7 @@ pub(crate) fn plan( slug: Some(map.slug.clone()), }); } - plan_topics_removals( - &mut topics, - &topics_automations, - &key, - existing, - topic_map, - &duplicate_topic_maps, - &functions_by_id, - )?; + plan_topics_removals(&mut topics, &key, &duplicate_topic_maps)?; facets.push(FacetMutation { template: facet.clone(), existing: existing.cloned(), @@ -285,12 +277,8 @@ pub(crate) fn plan( fn plan_topics_removals( topics: &mut BTreeMap, - _automations: &[&ProjectAutomation], selected_key: &str, - _facet: Option<&Function>, - _topic_map: Option<&Function>, duplicate_topic_maps: &[&Function], - _functions_by_id: &HashMap<&str, &Function>, ) -> Result<()> { let selected = topics .get_mut(selected_key) From 3e06d2cd5bafc6dd27400435d00596ac0e586a6a Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Thu, 15 Oct 2026 10:35:03 -0700 Subject: [PATCH 5/7] fix: select exported Topics settings by bound automation ID --- src/observability/pull.rs | 2 +- src/observability/template.rs | 187 +++++++++++++++++++++++++++------- 2 files changed, 151 insertions(+), 38 deletions(-) diff --git a/src/observability/pull.rs b/src/observability/pull.rs index 1058ea8a..1d3366d9 100644 --- a/src/observability/pull.rs +++ b/src/observability/pull.rs @@ -41,7 +41,7 @@ pub(crate) async fn run(base: BaseArgs, args: PullArgs) -> Result<()> { (template.facets, template.automations) = filter_active_resources(template.facets, template.automations); } - retain_topics_settings(&mut template); + retain_topics_settings(&mut template, &functions, &automations)?; validate(&template)?; validate_topics_scope(&template)?; // Selection happens first so a selected facet never loses its required definition. diff --git a/src/observability/template.rs b/src/observability/template.rs index 5ad30bed..dc7ab1d6 100644 --- a/src/observability/template.rs +++ b/src/observability/template.rs @@ -96,28 +96,19 @@ pub(crate) fn from_remote( let mut facet_templates = facets .into_iter() - .map(|facet| facet_from_remote(facet, topics.get(&facet.id).cloned(), &by_id)) + .map(|facet| { + facet_from_remote( + facet, + topics + .get(&facet.id) + .map(|automation| automation.name.clone()), + &by_id, + ) + }) .collect::>>()?; facet_templates.sort_by(|a, b| a.name.cmp(&b.name).then(a.slug.cmp(&b.slug))); - let names = facet_templates - .iter() - .filter_map(|facet| facet.topics_automation.as_deref()) - .collect::>(); - let mut topics_templates = automations - .iter() - .filter(|automation| { - is_topics(&automation.config) && names.contains(automation.name.as_str()) - }) - .map(|automation| { - Ok(AutomationTemplate { - name: automation.name.clone(), - description: automation.description.clone(), - config: topics_settings(&automation.config)?, - }) - }) - .collect::>>()?; - topics_templates.sort_by(|a, b| a.name.cmp(&b.name)); + let topics_templates = bundled_topics_settings(topics.into_values())?; let mut loop_templates = automations .iter() @@ -202,12 +193,12 @@ fn saved_preprocessor( Ok(Some(portable)) } -fn topics_by_facet( +fn topics_by_facet<'a>( facets: &[&Function], by_id: &HashMap<&str, &Function>, - automations: &[ProjectAutomation], -) -> Result> { - let mut names: HashMap> = HashMap::new(); + automations: &'a [ProjectAutomation], +) -> Result> { + let mut bindings: HashMap> = HashMap::new(); for automation in automations .iter() .filter(|automation| is_topics(&automation.config)) @@ -231,30 +222,51 @@ fn topics_by_facet( .copied() .filter(|facet| topic_map_matches(topic_map, facet)) { - names + bindings .entry(facet.id.clone()) .or_default() - .insert(automation.name.clone()); + .insert(automation.id.as_str(), automation); } } } - names + bindings .into_iter() - .map(|(facet_id, names)| { - if names.len() != 1 { - let mut names = names.into_iter().collect::>(); + .map(|(facet_id, automations)| { + if automations.len() != 1 { + let mut names = automations + .values() + .map(|automation| format!("{} ({})", automation.name, automation.id)) + .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"))) + Ok((facet_id, automations.into_values().next().expect("one automation"))) }) .collect() } +fn bundled_topics_settings<'a>( + automations: impl Iterator, +) -> Result> { + let mut ids = HashSet::new(); + let mut settings = automations + .filter(|automation| ids.insert(automation.id.as_str())) + .map(|automation| { + Ok(AutomationTemplate { + name: automation.name.clone(), + description: automation.description.clone(), + config: topics_settings(&automation.config)?, + }) + }) + .collect::>>()?; + settings.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(settings) +} + pub(crate) fn topic_map_matches(topic_map: &Function, facet: &Function) -> bool { let Some(data) = topic_map.function_data.as_ref() else { return false; @@ -406,15 +418,30 @@ pub(crate) fn validate_topics_scope(template: &ActiveObservabilityTemplate) -> R Ok(()) } -pub(crate) fn retain_topics_settings(template: &mut ActiveObservabilityTemplate) { - let names = template +pub(crate) fn retain_topics_settings( + template: &mut ActiveObservabilityTemplate, + functions: &[Function], + automations: &[ProjectAutomation], +) -> Result<()> { + let slugs = template .facets .iter() - .filter_map(|facet| facet.topics_automation.as_deref()) + .map(|facet| facet.slug.as_str()) .collect::>(); - template - .topics_automations - .retain(|automation| names.contains(automation.name.as_str())); + let facets = functions + .iter() + .filter(|function| { + function.function_type.as_deref() == Some("facet") + && slugs.contains(function.slug.as_str()) + }) + .collect::>(); + let by_id = functions + .iter() + .map(|function| (function.id.as_str(), function)) + .collect::>(); + let topics = topics_by_facet(&facets, &by_id, automations)?; + template.topics_automations = bundled_topics_settings(topics.into_values())?; + Ok(()) } pub(crate) fn topics_settings(config: &Value) -> Result { @@ -925,6 +952,92 @@ mod tests { assert_eq!(twice["topic_map_functions"].as_array().unwrap().len(), 1); } + #[test] + fn active_observability_pull_uses_bound_automation_id() { + let facet = function( + "fn-test-facet", + "test-facet", + "facet", + json!({"type": "facet"}), + ); + let map = function( + "fn-test-map", + "test-map", + "classifier", + json!({"type": "topic_map", "source_facet_function": {"type": "function", "id": facet.id}}), + ); + let mut bound = automation( + "Test Topics", + json!({ + "event_type": "topic", "scope": {"type": "span"}, + "topic_map_functions": [ + {"function": {"type": "function", "id": map.id}}, + {"function": {"type": "function", "id": map.id}} + ] + }), + ); + bound.id = "auto-test-bound".to_string(); + bound.description = Some("Test bound settings".to_string()); + let mut unrelated = automation( + "Test Topics", + json!({"event_type": "topic", "scope": {"type": "trace"}}), + ); + unrelated.id = "auto-test-unrelated".to_string(); + + for automations in [ + vec![bound.clone(), unrelated.clone()], + vec![unrelated.clone(), bound.clone()], + ] { + let mut template = from_remote(&[facet.clone(), map.clone()], &automations).unwrap(); + retain_topics_settings(&mut template, &[facet.clone(), map.clone()], &automations) + .unwrap(); + validate(&template).expect("an unrelated same-name automation must not block pull"); + assert_eq!(template.topics_automations.len(), 1); + assert_eq!(template.topics_automations[0].name, bound.name); + assert_eq!( + template.topics_automations[0].description, + bound.description + ); + assert_eq!( + template.topics_automations[0].config, + topics_settings(&bound.config).unwrap() + ); + assert_eq!( + template.facets[0].topics_automation.as_deref(), + Some("Test Topics") + ); + } + + let mut other_facet = facet.clone(); + other_facet.id = "fn-test-other-facet".to_string(); + other_facet.slug = "test-other-facet".to_string(); + let mut other_map = map.clone(); + other_map.id = "fn-test-other-map".to_string(); + other_map.function_data = Some( + json!({"type": "topic_map", "source_facet_function": {"type": "function", "id": other_facet.id}}), + ); + unrelated.config["topic_map_functions"] = + json!([{"function": {"type": "function", "id": other_map.id}}]); + let functions = vec![facet.clone(), map.clone(), other_facet, other_map]; + let automations = vec![unrelated.clone(), bound.clone()]; + let mut template = from_remote(&functions, &automations).unwrap(); + assert_eq!(template.topics_automations.len(), 2); + template + .facets + .retain(|selected| selected.slug == facet.slug); + retain_topics_settings(&mut template, &functions, &automations).unwrap(); + validate(&template).unwrap(); + assert_eq!(template.topics_automations.len(), 1); + assert_eq!( + template.topics_automations[0].description, + bound.description + ); + + unrelated.config = bound.config.clone(); + let error = from_remote(&[facet, map], &[bound, unrelated]).unwrap_err(); + assert!(error.to_string().contains("multiple Topics automations")); + } + #[test] fn active_observability_v2_settings_validate_and_follow_selection() { let mut source: ActiveObservabilityTemplate = serde_json::from_value(json!({ @@ -970,7 +1083,7 @@ mod tests { let error = validate_topics_scope(&source).unwrap_err().to_string(); assert!(error.contains("Test facet") && error.contains("specific experiment")); source.facets.clear(); - retain_topics_settings(&mut source); + retain_topics_settings(&mut source, &[], &[]).unwrap(); assert!(source.topics_automations.is_empty()); } } From e1f7b9d909451dac8f14238ecb2d9ba5e8d15fbd Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Thu, 15 Oct 2026 17:22:42 -0700 Subject: [PATCH 6/7] fix: use awaited vite-node entry for ESM eval retries Declare vite-node in both ESM retry fixtures so Node 20 does not depend on an on-demand npm install. --- src/eval.rs | 3 ++- tests/evals/js/eval-ts-esm-only-dep/package.json | 3 ++- tests/evals/js/eval-ts-local-sdk/package.json | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/eval.rs b/src/eval.rs index 91eae7c3..5b5770fc 100644 --- a/src/eval.rs +++ b/src/eval.rs @@ -2734,11 +2734,12 @@ fn build_js_plan( } fn build_vite_node_fallback_command(runner: &Path, files: &[String]) -> Result { + let runner = js_runner_path_for_kind(runner, RunnerKind::ViteNode); if let Some(path) = find_node_module_bin_for_files("vite-node", files) .or_else(|| js_runner::find_binary_in_path(&["vite-node"])) { let mut command = Command::new(path); - command.arg(runner).args(files); + command.arg(&runner).args(files); return Ok(command); } diff --git a/tests/evals/js/eval-ts-esm-only-dep/package.json b/tests/evals/js/eval-ts-esm-only-dep/package.json index 779b5d64..f4647432 100644 --- a/tests/evals/js/eval-ts-esm-only-dep/package.json +++ b/tests/evals/js/eval-ts-esm-only-dep/package.json @@ -6,6 +6,7 @@ "esm-only-pkg": "file:./esm-only-pkg" }, "devDependencies": { - "tsx": "^4.23.15" + "tsx": "^4.23.15", + "vite-node": "^6.0.0" } } diff --git a/tests/evals/js/eval-ts-local-sdk/package.json b/tests/evals/js/eval-ts-local-sdk/package.json index 6f2a4a11..8f47b168 100644 --- a/tests/evals/js/eval-ts-local-sdk/package.json +++ b/tests/evals/js/eval-ts-local-sdk/package.json @@ -7,6 +7,7 @@ "esm-only-pkg": "file:esm-only-pkg" }, "devDependencies": { - "tsx": "^4.23.15" + "tsx": "^4.23.15", + "vite-node": "^6.0.0" } } From 32442e6a983a8af96908f0e15d5dabdc8b8f48c1 Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Thu, 15 Oct 2026 17:22:42 -0700 Subject: [PATCH 7/7] fix: use awaited vite-node entry for ESM eval retries Declare vite-node in both ESM retry fixtures so Node 20 does not depend on an on-demand npm install. --- src/eval.rs | 3 ++- tests/evals/js/eval-ts-esm-only-dep/package.json | 3 ++- tests/evals/js/eval-ts-local-sdk/package.json | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/eval.rs b/src/eval.rs index 91eae7c3..5b5770fc 100644 --- a/src/eval.rs +++ b/src/eval.rs @@ -2734,11 +2734,12 @@ fn build_js_plan( } fn build_vite_node_fallback_command(runner: &Path, files: &[String]) -> Result { + let runner = js_runner_path_for_kind(runner, RunnerKind::ViteNode); if let Some(path) = find_node_module_bin_for_files("vite-node", files) .or_else(|| js_runner::find_binary_in_path(&["vite-node"])) { let mut command = Command::new(path); - command.arg(runner).args(files); + command.arg(&runner).args(files); return Ok(command); } diff --git a/tests/evals/js/eval-ts-esm-only-dep/package.json b/tests/evals/js/eval-ts-esm-only-dep/package.json index 779b5d64..f4647432 100644 --- a/tests/evals/js/eval-ts-esm-only-dep/package.json +++ b/tests/evals/js/eval-ts-esm-only-dep/package.json @@ -6,6 +6,7 @@ "esm-only-pkg": "file:./esm-only-pkg" }, "devDependencies": { - "tsx": "^4.23.15" + "tsx": "^4.23.15", + "vite-node": "^6.0.0" } } diff --git a/tests/evals/js/eval-ts-local-sdk/package.json b/tests/evals/js/eval-ts-local-sdk/package.json index 6f2a4a11..8f47b168 100644 --- a/tests/evals/js/eval-ts-local-sdk/package.json +++ b/tests/evals/js/eval-ts-local-sdk/package.json @@ -7,6 +7,7 @@ "esm-only-pkg": "file:esm-only-pkg" }, "devDependencies": { - "tsx": "^4.23.15" + "tsx": "^4.23.15", + "vite-node": "^6.0.0" } }