feat(observability): add core OTLP log and metric pipelines

crates/cli/src/plugins/editor_model.rs

@@ -1020,14 +1020,7 @@ pub(super) fn merge_known_editor_object(
 }
 
 pub(super) fn observability_editor_fields_with_version() -> Vec<&'static str> {
-    let mut keys = vec!["version"];
-    keys.extend(
-        ObservabilityConfig::editor_schema()
-            .fields
-            .iter()
-            .map(|field| field.name),
-    );
-    keys
+    nested_editor_keys(ObservabilityConfig::editor_schema())
 }
 
 pub(super) fn nested_editor_keys(

crates/cli/src/plugins/mod.rs

@@ -758,6 +758,35 @@ fn parse_float_value(field: &EditorFieldSpec, value: &str) -> Result<Value, CliE
     Ok(json!(parsed))
 }
 
+fn editor_enum_default_index(field: &EditorFieldSpec, current: Option<&Value>) -> usize {
+    current
+        .map(display_value)
+        .and_then(|value| {
+            field
+                .enum_values
+                .iter()
+                .position(|candidate| *candidate == value)
+        })
+        .unwrap_or(0)
+}
+
+fn editor_enum_value(field: &EditorFieldSpec, selected: usize) -> Value {
+    let value = field.enum_values[selected];
+    if field
+        .enum_values
+        .iter()
+        .all(|candidate| candidate.parse::<i64>().is_ok())
+    {
+        json!(
+            value
+                .parse::<i64>()
+                .expect("numeric editor enum values were validated above")
+        )
+    } else {
+        json!(value)
+    }
+}
+
 #[cfg(test)]
 #[path = "../../tests/coverage/shared/plugins_tests.rs"]
 mod tests;

crates/cli/src/plugins/prompt.rs

@@ -1219,17 +1219,14 @@ fn prompt_value(
         }
         EditorFieldKind::Enum => {
             let values = field.enum_values;
-            let default_idx = current
-                .and_then(Value::as_str)
-                .and_then(|value| values.iter().position(|candidate| *candidate == value))
-                .unwrap_or(0);
+            let default_idx = editor_enum_default_index(field, current);
             let idx = Select::with_theme(theme)
                 .with_prompt(field.label)
                 .items(values)
                 .default(default_idx)
                 .interact()
                 .map_err(editor_error)?;
-            Ok(json!(values[idx]))
+            Ok(editor_enum_value(field, idx))
         }
         EditorFieldKind::String => {
             let initial = current.and_then(Value::as_str).unwrap_or_default();

crates/cli/tests/coverage/shared/plugins_tests.rs

@@ -199,6 +199,28 @@ fn editor_uses_explicit_plugin_target_without_default_discovery() {
     assert_eq!(target, path);
 }
 
+fn assert_observability_signal_editor_sections(opentelemetry: &EditorSchema) {
+    for signal in ["logs", "metrics"] {
+        let signal = opentelemetry.field(signal).unwrap();
+        assert_eq!(signal.kind, EditorFieldKind::Section);
+        assert!(signal.optional);
+        let signal_schema = signal.schema().unwrap();
+        let endpoints = signal_schema.field("endpoints").unwrap();
+        assert_eq!(endpoints.kind, EditorFieldKind::List);
+        assert!(endpoints.optional);
+        let endpoint_schema = endpoints.list_item.unwrap().schema.unwrap()();
+        assert!(endpoint_schema.field("type").is_none());
+        assert_eq!(
+            endpoint_schema.field("endpoint").unwrap().kind,
+            EditorFieldKind::String
+        );
+        assert_eq!(
+            endpoint_schema.field("header_env").unwrap().kind,
+            EditorFieldKind::StringMap
+        );
+    }
+}
+
 #[test]
 fn typed_editor_model_contains_observability_sections() {
     let schema = ObservabilityConfig::editor_schema();
@@ -242,6 +264,29 @@ fn typed_editor_model_contains_observability_sections() {
             .iter()
             .any(|field| field.name == "header_env")
     );
+    assert_observability_signal_editor_sections(opentelemetry);
+    assert_eq!(
+        opentelemetry
+            .field("logs")
+            .unwrap()
+            .schema()
+            .unwrap()
+            .field("minimum_severity")
+            .unwrap()
+            .enum_values,
+        &["trace", "debug", "info", "warn", "warning", "error"]
+    );
+    assert_eq!(
+        opentelemetry
+            .field("metrics")
+            .unwrap()
+            .schema()
+            .unwrap()
+            .field("temporality")
+            .unwrap()
+            .enum_values,
+        &["cumulative", "delta", "low_memory"]
+    );
 }
 
 #[test]
@@ -516,10 +561,12 @@ fn component_menu_contains_toggle_fields_and_back() {
         .collect::<Vec<_>>();
 
     assert_eq!(labels[0], "Toggle component [on]");
-    assert_eq!(labels[1], "  Edit ATOF");
+    assert_eq!(labels[1], "✓ Edit version");
+    assert_eq!(labels[2], "  Edit ATOF");
     assert_eq!(labels.last().unwrap(), "Back [q]");
     assert!(matches!(actions[0], ComponentMenuAction::Toggle));
     assert!(matches!(actions[1], ComponentMenuAction::EditField(0)));
+    assert!(matches!(actions[2], ComponentMenuAction::EditField(1)));
     assert!(matches!(actions.last(), Some(ComponentMenuAction::Back)));
 }
 
@@ -823,6 +870,15 @@ fn editor_model_object_and_schema_helpers_cover_fallbacks() {
 
     let fields = observability_editor_fields_with_version();
     assert_eq!(fields.first(), Some(&"version"));
+    assert_eq!(
+        fields.iter().filter(|field| **field == "version").count(),
+        1
+    );
+    let version = ObservabilityConfig::editor_schema()
+        .field("version")
+        .expect("observability version field");
+    assert_eq!(version.kind, EditorFieldKind::Enum);
+    assert_eq!(version.enum_values, &["3", "4"]);
     let nested = nested_editor_keys(ObservabilityConfig::editor_schema());
     assert!(nested.contains(&"atof"));
 
@@ -1054,6 +1110,7 @@ fn editor_save_persists_disabled_nemo_guardrails_policy_only_edits() {
 #[test]
 fn typed_editor_serializes_explicit_observability_overrides() {
     let mut observability = ObservabilityConfig::default();
+    assert_eq!(observability.version, 4);
     let atof = ObservabilityConfig::editor_schema().field("atof").unwrap();
     toggle_section(&mut observability, atof);
     set_section_field(
@@ -1065,6 +1122,7 @@ fn typed_editor_serializes_explicit_observability_overrides() {
     .unwrap();
 
     let map = observability_config_map(&observability).unwrap();
+    assert_eq!(map.get("version"), Some(&json!(4)));
     let atof = map
         .get("atof")
         .and_then(Value::as_object)
@@ -2963,6 +3021,30 @@ fn parse_float_value_rejects_non_finite_numbers() {
     }
 }
 
+#[test]
+fn editor_enum_values_preserve_numeric_config_fields() {
+    let version = ObservabilityConfig::editor_schema()
+        .field("version")
+        .expect("observability version field");
+    assert_eq!(editor_enum_default_index(&version, Some(&json!(4))), 1);
+    assert_eq!(editor_enum_value(&version, 0), json!(3));
+    assert_eq!(editor_enum_value(&version, 1), json!(4));
+
+    let severity = ObservabilityConfig::editor_schema()
+        .field("opentelemetry")
+        .and_then(EditorFieldSpec::schema)
+        .and_then(|opentelemetry| opentelemetry.field("logs"))
+        .and_then(EditorFieldSpec::schema)
+        .and_then(|logs| logs.field("minimum_severity"))
+        .expect("minimum severity field");
+    let warning = severity
+        .enum_values
+        .iter()
+        .position(|value| *value == "warning")
+        .expect("warning alias");
+    assert_eq!(editor_enum_value(&severity, warning), json!("warning"));
+}
+
 fn tagged_list_item_schema() -> &'static EditorSchema {
     static SCHEMA: EditorSchema = EditorSchema { fields: &[] };
     &SCHEMA

crates/core/Cargo.toml

@@ -64,14 +64,14 @@ tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-t
 tokio-stream = { version = "0.1", default-features = false, features = ["sync"] }
 typed-builder = "0.23.2"
 futures-util = "0.3"
-opentelemetry = { workspace = true, features = ["trace"] }
+opentelemetry = { workspace = true, features = ["trace", "logs", "metrics"] }
 opentelemetry-semantic-conventions.workspace = true
-opentelemetry_sdk = { workspace = true, features = ["trace", "internal-logs"] }
+opentelemetry_sdk = { workspace = true, features = ["trace", "logs", "metrics", "internal-logs"] }
 tracing = { version = "0.1", default-features = false, features = ["std", "log"] }
 libloading = "0.8"
 semver = "1"
 sha2 = "0.11"
-opentelemetry-otlp = { version = "0.32", default-features = false, features = ["trace", "http-proto", "reqwest-blocking-client", "reqwest-rustls", "grpc-tonic", "tls-ring", "tls-roots"] }
+opentelemetry-otlp = { version = "0.32", default-features = false, features = ["trace", "logs", "metrics", "http-proto", "reqwest-blocking-client", "reqwest-rustls", "grpc-tonic", "tls-ring", "tls-roots"] }
 reqwest = { version = "0.12", default-features = false, features = ["charset", "http2", "rustls-tls-native-roots", "stream"] }
 tonic = { version = "0.14.1", default-features = false }
 object_store = { version = "0.14.1", default-features = false, features = ["aws"], optional = true }
@@ -86,7 +86,11 @@ unicode-general-category = "1.1"
 [dev-dependencies]
 tokio = { version = "1", features = ["rt", "macros", "sync", "test-util", "rt-multi-thread", "time"] }
 futures = "0.3"
-opentelemetry_sdk = { workspace = true, features = ["trace", "testing"] }
+opentelemetry_sdk = { workspace = true, features = ["trace", "logs", "metrics", "testing"] }
+opentelemetry-proto = { version = "0.32", default-features = false, features = ["gen-tonic", "logs", "metrics"] }
+prost = "0.14"
+tonic = { version = "0.14.1", features = ["transport"] }
+tokio-stream = { version = "0.1", features = ["net"] }
 serde_json = "1"
 object_store = { version = "0.14.1", default-features = false, features = ["aws"] }
 tempfile = "3"

crates/core/src/api/llm.rs

@@ -32,7 +32,7 @@ use crate::api::runtime::{
 };
 use crate::api::runtime::{ScopeStackHandle, capture_traceparent, current_scope_stack};
 use crate::api::scope::event;
-use crate::api::scope::{EmitMarkEventParams, ScopeHandle};
+use crate::api::scope::{EmitMarkEventParams, ScopeHandle, metadata_with_log_severity};
 use crate::api::shared::{
     ensure_runtime_owner, inject_dynamo_session_ids, inject_traceparent, inject_traceparent_value,
     metadata_with_otel_error, metadata_with_otel_status, resolve_parent_uuid,
@@ -629,13 +629,15 @@ async fn emit_pending_request_marks(
     ensure_runtime_owner()?;
     let timestamp = handle.started_at + TimeDelta::microseconds(1);
     for mark in marks {
+        let metadata = metadata_with_log_severity(mark.metadata, mark.severity)?;
         let event = Event::Mark(MarkEvent::new(
             BaseEvent::builder()
                 .name(mark.name)
                 .parent_uuid(handle.uuid)
                 .timestamp(timestamp)
                 .data_opt(mark.data)
-                .metadata_opt(mark.metadata)
+                .data_schema_opt(mark.data_schema)
+                .metadata_opt(metadata)
                 .build(),
             mark.category,
             mark.category_profile,