diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 73affc703ef8..a665c38df15a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4004,7 +4004,6 @@ dependencies = [ "codex-utils-absolute-path", "codex-utils-path-uri", "codex-utils-string", - "opentelemetry_sdk", "pretty_assertions", "schemars 0.8.22", "serde", diff --git a/codex-rs/ext/skills/Cargo.toml b/codex-rs/ext/skills/Cargo.toml index df32e02f39ca..c12bb8e1a384 100644 --- a/codex-rs/ext/skills/Cargo.toml +++ b/codex-rs/ext/skills/Cargo.toml @@ -31,6 +31,5 @@ url = { workspace = true } [dev-dependencies] codex-utils-absolute-path = { workspace = true } -opentelemetry_sdk = { workspace = true } pretty_assertions = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/codex-rs/ext/skills/src/dynamic_skill_selector.rs b/codex-rs/ext/skills/src/dynamic_skill_selector.rs index 948c815f1a53..fdb9531006d9 100644 --- a/codex-rs/ext/skills/src/dynamic_skill_selector.rs +++ b/codex-rs/ext/skills/src/dynamic_skill_selector.rs @@ -1,4 +1,6 @@ +mod fielded_bm25; mod weighted_lexical; +pub(crate) use fielded_bm25::FieldedBm25SkillSelector; pub(crate) use weighted_lexical::WeightedLexicalSkillSelector; /// Metadata searched by a cheap skill selector. diff --git a/codex-rs/ext/skills/src/dynamic_skill_selector/fielded_bm25.rs b/codex-rs/ext/skills/src/dynamic_skill_selector/fielded_bm25.rs new file mode 100644 index 000000000000..11b6eddb1191 --- /dev/null +++ b/codex-rs/ext/skills/src/dynamic_skill_selector/fielded_bm25.rs @@ -0,0 +1,239 @@ +use std::collections::HashMap; +use std::collections::HashSet; + +use super::CheapSkillSelection; +use super::CheapSkillSelector; +use super::SkillSelectionDocument; + +const MAX_QUERY_BYTES: usize = 4 * 1024; +const MAX_QUERY_TERMS: usize = 64; +const MAX_DOCUMENT_BYTES: usize = 4 * 1024; +const MAX_DOCUMENT_TERMS: usize = 256; +const MAX_CANDIDATES: usize = 1_000; +const MAX_RESULTS: usize = 50; +const FIELD_WEIGHTS: [f64; 3] = [8.0, 4.0, 1.0]; +const K1: f64 = 1.2; +const B: f64 = 0.75; + +const STOP_WORDS: &[&str] = &[ + "a", "an", "and", "are", "as", "at", "be", "by", "do", "for", "from", "how", "i", "in", "is", + "it", "me", "my", "of", "on", "or", "please", "that", "the", "this", "to", "use", "we", "what", + "when", "where", "which", "with", "you", "your", +]; + +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct FieldedBm25SkillSelector; + +impl CheapSkillSelector for FieldedBm25SkillSelector { + fn method(&self) -> &'static str { + "fielded_bm25_v1" + } + + fn select( + &self, + query: &str, + documents: &[SkillSelectionDocument<'_>], + limit: usize, + ) -> CheapSkillSelection { + let (query, query_bytes_truncated) = bounded(query, MAX_QUERY_BYTES); + let (query_terms, query_terms_truncated) = query_terms(query); + let query_truncated = query_bytes_truncated || query_terms_truncated; + let candidate_set_truncated = documents.len() > MAX_CANDIDATES; + if query_terms.is_empty() || limit == 0 { + return CheapSkillSelection { + query_term_count: query_terms.len(), + query_truncated, + candidate_set_truncated, + ..Default::default() + }; + } + + let prepared = documents + .iter() + .take(MAX_CANDIDATES) + .map(PreparedDocument::new) + .collect::>(); + let averages = average_field_lengths(&prepared); + let document_frequencies = document_frequencies(&prepared); + let document_count = prepared.len() as f64; + let mut scored = prepared + .iter() + .filter_map(|document| { + let score = score_document( + document, + &query_terms, + &document_frequencies, + document_count, + averages, + ); + (score > 0.0).then_some((score, document.id, document.name)) + }) + .collect::>(); + scored.sort_by(|left, right| { + right + .0 + .total_cmp(&left.0) + .then_with(|| left.2.cmp(right.2)) + .then_with(|| left.1.cmp(&right.1)) + }); + + CheapSkillSelection { + candidate_ids: scored + .into_iter() + .take(limit.min(MAX_RESULTS)) + .map(|(_, id, _)| id) + .collect(), + query_term_count: query_terms.len(), + query_truncated, + candidate_set_truncated, + } + } +} + +struct PreparedDocument<'a> { + id: usize, + name: &'a str, + fields: [Vec; 3], +} + +impl<'a> PreparedDocument<'a> { + fn new(document: &'a SkillSelectionDocument<'a>) -> Self { + Self { + id: document.id, + name: document.name, + fields: [ + document_terms(document.name), + document_terms(document.short_description.unwrap_or_default()), + document_terms(document.description), + ], + } + } +} + +fn score_document( + document: &PreparedDocument<'_>, + query_terms: &[String], + document_frequencies: &HashMap, + document_count: f64, + average_field_lengths: [f64; 3], +) -> f64 { + query_terms.iter().fold(0.0, |score, query_term| { + let frequency = document_frequencies + .get(query_term) + .copied() + .unwrap_or_default() as f64; + if frequency == 0.0 { + return score; + } + let weighted_term_frequency = + document + .fields + .iter() + .enumerate() + .fold(0.0, |weighted, (field_index, terms)| { + let term_frequency = + terms.iter().filter(|term| *term == query_term).count() as f64; + if term_frequency == 0.0 { + return weighted; + } + let average_length = average_field_lengths[field_index]; + let length_ratio = if average_length == 0.0 { + 1.0 + } else { + terms.len() as f64 / average_length + }; + weighted + + FIELD_WEIGHTS[field_index] * term_frequency / (1.0 - B + B * length_ratio) + }); + if weighted_term_frequency == 0.0 { + return score; + } + let inverse_document_frequency = + (1.0 + (document_count - frequency + 0.5) / (frequency + 0.5)).ln(); + score + + inverse_document_frequency * weighted_term_frequency * (K1 + 1.0) + / (weighted_term_frequency + K1) + }) +} + +fn average_field_lengths(documents: &[PreparedDocument<'_>]) -> [f64; 3] { + if documents.is_empty() { + return [0.0; 3]; + } + let totals = documents.iter().fold([0usize; 3], |mut totals, document| { + for (index, field) in document.fields.iter().enumerate() { + totals[index] = totals[index].saturating_add(field.len()); + } + totals + }); + totals.map(|total| total as f64 / documents.len() as f64) +} + +fn document_frequencies(documents: &[PreparedDocument<'_>]) -> HashMap { + let mut frequencies = HashMap::new(); + for document in documents { + let terms = document + .fields + .iter() + .flatten() + .map(String::as_str) + .collect::>(); + for term in terms { + *frequencies.entry(term.to_string()).or_default() += 1; + } + } + frequencies +} + +fn query_terms(query: &str) -> (Vec, bool) { + let mut seen = HashSet::new(); + let mut terms = Vec::new(); + for term in normalized_terms(query) + .into_iter() + .filter(|term| term.chars().count() >= 2 && !STOP_WORDS.contains(&term.as_str())) + { + if !seen.insert(term.clone()) { + continue; + } + if terms.len() == MAX_QUERY_TERMS { + return (terms, true); + } + terms.push(term); + } + (terms, false) +} + +fn document_terms(value: &str) -> Vec { + let (value, _) = bounded(value, MAX_DOCUMENT_BYTES); + normalized_terms(value) + .into_iter() + .take(MAX_DOCUMENT_TERMS) + .collect() +} + +fn normalized_terms(value: &str) -> Vec { + let mut normalized = String::with_capacity(value.len()); + for character in value.chars() { + if character.is_alphanumeric() { + normalized.extend(character.to_lowercase()); + } else { + normalized.push(' '); + } + } + normalized.split_whitespace().map(str::to_string).collect() +} + +fn bounded(value: &str, max_bytes: usize) -> (&str, bool) { + if value.len() <= max_bytes { + return (value, false); + } + let mut end = max_bytes; + while !value.is_char_boundary(end) { + end = end.saturating_sub(1); + } + (&value[..end], true) +} + +#[cfg(test)] +#[path = "fielded_bm25_tests.rs"] +mod tests; diff --git a/codex-rs/ext/skills/src/dynamic_skill_selector/fielded_bm25_tests.rs b/codex-rs/ext/skills/src/dynamic_skill_selector/fielded_bm25_tests.rs new file mode 100644 index 000000000000..14bebaf7c38c --- /dev/null +++ b/codex-rs/ext/skills/src/dynamic_skill_selector/fielded_bm25_tests.rs @@ -0,0 +1,79 @@ +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn bm25_prioritizes_rare_terms() { + let documents = [ + document(/*id*/ 1, "review-helper", "Review code and prose."), + document( + /*id*/ 2, + "terraform-review", + "Review Terraform infrastructure.", + ), + document(/*id*/ 3, "document-review", "Review Word documents."), + ]; + + let selection = + FieldedBm25SkillSelector.select("review terraform", &documents, /*limit*/ 20); + + assert_eq!(vec![2, 3, 1], selection.candidate_ids); +} + +#[test] +fn bm25_weights_names_above_descriptions() { + let documents = [ + document(/*id*/ 1, "slides", "Create presentations."), + document(/*id*/ 2, "presentations", "Create and edit slides."), + ]; + + let selection = FieldedBm25SkillSelector.select("slides", &documents, /*limit*/ 20); + + assert_eq!(vec![1, 2], selection.candidate_ids); +} + +#[test] +fn bm25_drops_candidates_without_matching_terms() { + let documents = [document( + /*id*/ 1, + "spreadsheets", + "Analyze tabular data.", + )]; + + let selection = + FieldedBm25SkillSelector.select("render a video", &documents, /*limit*/ 20); + + assert!(selection.candidate_ids.is_empty()); +} + +#[test] +fn bm25_reports_bounded_inputs() { + let long_query = "match ".repeat(MAX_QUERY_BYTES); + let names = (0..=MAX_CANDIDATES) + .map(|index| format!("candidate-{index}")) + .collect::>(); + let documents = names + .iter() + .enumerate() + .map(|(id, name)| SkillSelectionDocument { + id, + name, + short_description: None, + description: "match", + }) + .collect::>(); + + let selection = FieldedBm25SkillSelector.select(&long_query, &documents, /*limit*/ 20); + + assert!(selection.query_truncated); + assert!(selection.candidate_set_truncated); + assert_eq!(20, selection.candidate_ids.len()); +} + +fn document<'a>(id: usize, name: &'a str, description: &'a str) -> SkillSelectionDocument<'a> { + SkillSelectionDocument { + id, + name, + short_description: None, + description, + } +} diff --git a/codex-rs/ext/skills/src/shadow_selection_experiment.rs b/codex-rs/ext/skills/src/shadow_selection_experiment.rs index a6147046389f..b2fba0d1e36e 100644 --- a/codex-rs/ext/skills/src/shadow_selection_experiment.rs +++ b/codex-rs/ext/skills/src/shadow_selection_experiment.rs @@ -13,6 +13,7 @@ use crate::catalog::SkillCatalog; use crate::catalog::SkillSourceKind; use crate::dynamic_skill_selector::CheapSkillSelection; use crate::dynamic_skill_selector::CheapSkillSelector; +use crate::dynamic_skill_selector::FieldedBm25SkillSelector; use crate::dynamic_skill_selector::SkillSelectionDocument; use crate::dynamic_skill_selector::WeightedLexicalSkillSelector; @@ -35,7 +36,10 @@ pub(crate) struct ShadowSelectionExperiment { impl ShadowSelectionExperiment { pub(crate) fn new(metrics_client: Option) -> Self { Self { - selectors: vec![Box::new(WeightedLexicalSkillSelector)], + selectors: vec![ + Box::new(WeightedLexicalSkillSelector), + Box::new(FieldedBm25SkillSelector), + ], metrics_client, } } diff --git a/codex-rs/ext/skills/tests/implicit_invocation.rs b/codex-rs/ext/skills/tests/implicit_invocation.rs deleted file mode 100644 index 001c5cc83bef..000000000000 --- a/codex-rs/ext/skills/tests/implicit_invocation.rs +++ /dev/null @@ -1,391 +0,0 @@ -use std::collections::BTreeMap; -use std::sync::Arc; - -use codex_core_skills::HostSkillsSnapshot; -use codex_core_skills::SkillLoadOutcome; -use codex_core_skills::SkillMetadata; -use codex_extension_api::ConversationHistory; -use codex_extension_api::ExtensionData; -use codex_extension_api::ExtensionRegistry; -use codex_extension_api::ExtensionRegistryBuilder; -use codex_extension_api::NoopTurnItemEmitter; -use codex_extension_api::SkillInvocationInput; -use codex_extension_api::SkillInvocationKind; -use codex_extension_api::ThreadStartInput; -use codex_extension_api::ToolCall; -use codex_extension_api::ToolPayload; -use codex_extension_api::TurnInputContext; -use codex_extension_api::WorldStateContributionInput; -use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; -use codex_otel::MetricsClient; -use codex_otel::MetricsConfig; -use codex_protocol::capabilities::CapabilityRootLocation; -use codex_protocol::capabilities::SelectedCapabilityRoot; -use codex_protocol::protocol::SessionSource; -use codex_protocol::protocol::SkillScope; -use codex_protocol::protocol::TruncationPolicy; -use codex_protocol::user_input::UserInput; -use codex_skills_extension::HostSkillProvider; -use codex_skills_extension::SkillProviders; -use codex_skills_extension::SkillsExtensionConfig; -use codex_skills_extension::catalog::SkillAuthority; -use codex_skills_extension::catalog::SkillCatalog; -use codex_skills_extension::catalog::SkillCatalogEntry; -use codex_skills_extension::catalog::SkillPackageId; -use codex_skills_extension::catalog::SkillReadResult; -use codex_skills_extension::catalog::SkillResourceId; -use codex_skills_extension::catalog::SkillSearchResult; -use codex_skills_extension::catalog::SkillSourceKind; -use codex_skills_extension::install_with_providers_and_metrics; -use codex_skills_extension::provider::SkillListQuery; -use codex_skills_extension::provider::SkillProvider; -use codex_skills_extension::provider::SkillProviderFuture; -use codex_skills_extension::provider::SkillReadRequest; -use codex_skills_extension::provider::SkillSearchRequest; -use codex_utils_absolute_path::AbsolutePathBuf; -use codex_utils_path_uri::PathUri; -use opentelemetry_sdk::metrics::InMemoryMetricExporter; -use opentelemetry_sdk::metrics::data::AggregatedMetrics; -use opentelemetry_sdk::metrics::data::MetricData; -use pretty_assertions::assert_eq; - -type TestResult = Result<(), Box>; -type InvocationPoint = (BTreeMap, u64); - -const INVOCATION_METRIC: &str = "codex.skills.shadow_selection.invocation"; -const PACKAGE: &str = "orchestrator/demo"; -const MAIN_RESOURCE: &str = "skill://orchestrator/demo/SKILL.md"; - -#[derive(Clone)] -struct OrchestratorProvider; - -impl SkillProvider for OrchestratorProvider { - fn list(&self, _query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> { - Box::pin(std::future::ready(Ok(SkillCatalog { - entries: vec![SkillCatalogEntry::new( - SkillPackageId(PACKAGE.to_string()), - SkillAuthority::new(SkillSourceKind::Orchestrator, CODEX_APPS_MCP_SERVER_NAME), - "demo", - "Use the demo skill.", - SkillResourceId::new(MAIN_RESOURCE), - )], - warnings: Vec::new(), - }))) - } - - fn read(&self, request: SkillReadRequest) -> SkillProviderFuture<'_, SkillReadResult> { - Box::pin(std::future::ready(Ok(SkillReadResult { - resource: request.resource, - contents: "# Demo\n\nUse the demo skill.".to_string(), - }))) - } - - fn search(&self, _request: SkillSearchRequest) -> SkillProviderFuture<'_, SkillSearchResult> { - Box::pin(std::future::ready(Ok(SkillSearchResult::default()))) - } -} - -#[derive(Clone)] -struct ExecutorProvider; - -impl SkillProvider for ExecutorProvider { - fn list(&self, _query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> { - Box::pin(std::future::ready(Ok(SkillCatalog { - entries: (0..5) - .map(|index| { - let resource = format!("skill://executor/demo-{index}/SKILL.md"); - SkillCatalogEntry::new( - SkillPackageId(format!("executor/demo-{index}")), - SkillAuthority::new(SkillSourceKind::Executor, "executor"), - "演示文稿", - "创建演示文稿。", - SkillResourceId::new(resource), - ) - }) - .collect(), - warnings: Vec::new(), - }))) - } - - fn read(&self, request: SkillReadRequest) -> SkillProviderFuture<'_, SkillReadResult> { - Box::pin(std::future::ready(Ok(SkillReadResult { - resource: request.resource, - contents: "# 演示文稿".to_string(), - }))) - } - - fn search(&self, _request: SkillSearchRequest) -> SkillProviderFuture<'_, SkillSearchResult> { - Box::pin(std::future::ready(Ok(SkillSearchResult::default()))) - } -} - -#[tokio::test] -async fn implicit_core_and_native_read_invocations_share_turn_local_recording() -> TestResult { - let metrics = MetricsClient::new( - MetricsConfig::in_memory( - "test", - "codex-skills-extension", - env!("CARGO_PKG_VERSION"), - InMemoryMetricExporter::default(), - ) - .with_runtime_reader(), - )?; - let mut builder = ExtensionRegistryBuilder::<()>::new(); - install_with_providers_and_metrics( - &mut builder, - SkillProviders::new().with_orchestrator_provider(Arc::new(OrchestratorProvider)), - Some(metrics.clone()), - |_| SkillsExtensionConfig { - include_instructions: false, - bundled_skills_enabled: false, - orchestrator_skills_enabled: true, - shadow_selection_enabled: true, - }, - ); - let registry = builder.build(); - let session_store = ExtensionData::new("session"); - let thread_store = ExtensionData::new("thread"); - registry.thread_lifecycle_contributors()[0] - .on_thread_start(ThreadStartInput { - config: &(), - session_source: &SessionSource::Cli, - persistent_thread_state_available: true, - environments: &[], - session_store: &session_store, - thread_store: &thread_store, - }) - .await; - - let core_turn = start_turn(®istry, &session_store, &thread_store, "turn-core").await; - registry.skill_invocation_contributors()[0] - .on_skill_invocation(SkillInvocationInput { - session_store: &session_store, - thread_store: &thread_store, - turn_store: core_turn.as_ref(), - turn_id: "turn-core", - skill_resource: MAIN_RESOURCE, - kind: SkillInvocationKind::Implicit, - }) - .await; - - start_turn(®istry, &session_store, &thread_store, "turn-tool").await; - let tools = registry.tool_contributors()[0].tools(&session_store, &thread_store); - let read_tool = tools - .iter() - .find(|tool| tool.tool_name().name == "read") - .ok_or("skills.read tool should be registered")?; - for call_id in ["call-1", "call-2"] { - read_tool - .handle(ToolCall { - turn_id: "turn-tool".to_string(), - call_id: call_id.to_string(), - tool_name: read_tool.tool_name(), - model: "gpt-test".to_string(), - codex_turn_metadata: None, - truncation_policy: TruncationPolicy::Bytes(1_024), - conversation_history: ConversationHistory::default(), - turn_item_emitter: Arc::new(NoopTurnItemEmitter), - environments: Vec::new(), - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "authority": {"kind": "orchestrator"}, - "package": PACKAGE, - "resource": MAIN_RESOURCE, - }) - .to_string(), - }, - }) - .await?; - } - - assert_eq!( - vec![( - BTreeMap::from([ - ("hit".to_string(), "true".to_string()), - ("method".to_string(), "weighted_lexical_v1".to_string()), - ("query_script".to_string(), "ascii_latin".to_string()), - ("rank".to_string(), "1".to_string()), - ]), - 2, - )], - invocation_points(&metrics)?, - ); - Ok(()) -} - -#[tokio::test] -async fn shadow_selection_uses_host_snapshot_and_excludes_executor_candidates() -> TestResult { - let metrics = MetricsClient::new( - MetricsConfig::in_memory( - "test", - "codex-skills-extension", - env!("CARGO_PKG_VERSION"), - InMemoryMetricExporter::default(), - ) - .with_runtime_reader(), - )?; - let mut builder = ExtensionRegistryBuilder::<()>::new(); - install_with_providers_and_metrics( - &mut builder, - SkillProviders::new() - .with_executor_provider(Arc::new(ExecutorProvider)) - .with_host_provider(Arc::new(HostSkillProvider::new())), - Some(metrics.clone()), - |_| SkillsExtensionConfig { - include_instructions: true, - bundled_skills_enabled: false, - orchestrator_skills_enabled: false, - shadow_selection_enabled: true, - }, - ); - let registry = builder.build(); - let session_store = ExtensionData::new("session"); - let thread_store = ExtensionData::new("thread"); - registry.thread_lifecycle_contributors()[0] - .on_thread_start(ThreadStartInput { - config: &(), - session_source: &SessionSource::Cli, - persistent_thread_state_available: true, - environments: &[], - session_store: &session_store, - thread_store: &thread_store, - }) - .await; - - let skill_path = AbsolutePathBuf::try_from( - std::env::temp_dir() - .join("codex-shadow-host-skill") - .join("SKILL.md"), - )?; - let skill_resource = skill_path.to_string_lossy().into_owned(); - let mut outcome = SkillLoadOutcome::default(); - outcome.skills.push(SkillMetadata { - name: "演示文稿".to_string(), - description: "创建演示文稿。".to_string(), - short_description: None, - interface: None, - dependencies: None, - policy: None, - path_to_skills_md: skill_path, - scope: SkillScope::User, - plugin_id: None, - }); - let turn_store = ExtensionData::new("turn-host"); - turn_store.insert(HostSkillsSnapshot::new(Arc::new(outcome))); - let selected_roots = [SelectedCapabilityRoot { - id: "executor-root".to_string(), - location: CapabilityRootLocation::Environment { - environment_id: "executor".to_string(), - path: PathUri::parse("file:///skills").expect("executor skill root URI"), - }, - }]; - registry.context_contributors()[0] - .contribute_world_state(WorldStateContributionInput { - thread_id: codex_protocol::ThreadId::new(), - turn_id: "turn-host", - environments: &[], - ready_selected_capability_roots: &selected_roots, - session_store: &session_store, - thread_store: &thread_store, - turn_store: &turn_store, - }) - .await; - - let fragments = registry.turn_input_contributors()[0] - .contribute( - TurnInputContext { - turn_id: "turn-host".to_string(), - user_input: vec![UserInput::Text { - text: "演示文稿".to_string(), - text_elements: Vec::new(), - }], - environments: Vec::new(), - }, - &session_store, - &thread_store, - &turn_store, - ) - .await; - assert!(fragments.is_empty()); - - registry.skill_invocation_contributors()[0] - .on_skill_invocation(SkillInvocationInput { - session_store: &session_store, - thread_store: &thread_store, - turn_store: &turn_store, - turn_id: "turn-host", - skill_resource: &skill_resource, - kind: SkillInvocationKind::Implicit, - }) - .await; - - assert_eq!( - vec![( - BTreeMap::from([ - ("hit".to_string(), "true".to_string()), - ("method".to_string(), "weighted_lexical_v1".to_string()), - ("query_script".to_string(), "cjk".to_string()), - ("rank".to_string(), "1".to_string()), - ]), - 1, - )], - invocation_points(&metrics)?, - ); - Ok(()) -} - -async fn start_turn( - registry: &ExtensionRegistry<()>, - session_store: &ExtensionData, - thread_store: &ExtensionData, - turn_id: &str, -) -> Arc { - let turn_store = Arc::new(ExtensionData::new(turn_id)); - registry.turn_input_contributors()[0] - .contribute( - TurnInputContext { - turn_id: turn_id.to_string(), - user_input: vec![UserInput::Text { - text: "use demo".to_string(), - text_elements: Vec::new(), - }], - environments: Vec::new(), - }, - session_store, - thread_store, - turn_store.as_ref(), - ) - .await; - turn_store -} - -fn invocation_points( - metrics: &MetricsClient, -) -> Result, Box> { - let snapshot = metrics.snapshot()?; - let metric = snapshot - .scope_metrics() - .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) - .find(|metric| metric.name() == INVOCATION_METRIC) - .ok_or("invocation metric was not emitted")?; - match metric.data() { - AggregatedMetrics::U64(MetricData::Sum(sum)) => Ok(sum - .data_points() - .map(|point| { - ( - point - .attributes() - .map(|attribute| { - ( - attribute.key.as_str().to_string(), - attribute.value.as_str().to_string(), - ) - }) - .collect(), - point.value(), - ) - }) - .collect()), - other => Err(format!("unexpected invocation metric data: {other:?}").into()), - } -}