From b615ce9291f5fa31b21287f3919ebc427d29fbe2 Mon Sep 17 00:00:00 2001 From: Albin Cassirer Date: Sun, 19 Apr 2026 15:17:38 -0700 Subject: [PATCH] [observability] Add explicit field use markers Add FieldUse metadata so fields declare the exact projections intended to consume them. Detail level and data class remain guardrails enforced by FieldPolicy, while uses drives selection and prevents broad policies from accidentally collecting every safe-looking field. Support struct-level default uses in the Observation derive, with field-level uses overriding the default for mixed events. AppUsed now declares analytics once at the event level instead of repeating it on every field. Keep analytics conformance focused on final TrackEventRequest equality and move marker/policy filtering coverage into codex-observability tests. Update the design doc to describe the intent-versus-guardrail split. --- .../analytics/src/analytics_client_tests.rs | 20 ++--- codex-rs/observability-derive/src/lib.rs | 63 +++++++++++++- codex-rs/observability/src/events.rs | 2 +- codex-rs/observability/src/lib.rs | 84 ++++++++++++++++++- codex-rs/observability/tests/derive.rs | 59 ++++++------- docs/observability-event-stream-design.md | 25 ++++-- 6 files changed, 196 insertions(+), 57 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index cce90cca8e..b765bfd2d4 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1560,6 +1560,14 @@ async fn app_used_observation_matches_legacy_analytics_fact() { let mut observation_reducer = AnalyticsObservationReducer::default(); let mut legacy_events = Vec::new(); let mut observation_events = Vec::new(); + let observation = codex_observability::events::AppUsed { + model_slug: "gpt-5", + thread_id: "thread-1", + turn_id: "turn-1", + connector_id: Some("drive"), + app_name: Some("Drive"), + invocation_type: Some(codex_observability::events::InvocationType::Implicit), + }; legacy_reducer .ingest( @@ -1580,17 +1588,7 @@ async fn app_used_observation_matches_legacy_analytics_fact() { .await; observation_reducer - .ingest_app_used( - codex_observability::events::AppUsed { - model_slug: "gpt-5", - thread_id: "thread-1", - turn_id: "turn-1", - connector_id: Some("drive"), - app_name: Some("Drive"), - invocation_type: Some(codex_observability::events::InvocationType::Implicit), - }, - &mut observation_events, - ) + .ingest_app_used(observation, &mut observation_events) .await; let legacy_payload = serde_json::to_value(&legacy_events).expect("serialize legacy events"); diff --git a/codex-rs/observability-derive/src/lib.rs b/codex-rs/observability-derive/src/lib.rs index 0f8412e02d..deab932396 100644 --- a/codex-rs/observability-derive/src/lib.rs +++ b/codex-rs/observability-derive/src/lib.rs @@ -11,7 +11,9 @@ use quote::quote; use syn::Data; use syn::DeriveInput; use syn::Expr; +use syn::ExprLit; use syn::Fields; +use syn::Lit; use syn::LitStr; use syn::Path; use syn::parse_macro_input; @@ -22,6 +24,8 @@ use syn::parse_macro_input; /// /// - `#[observation(name = "domain.event")]` on the struct. /// - `#[obs(level = "basic|detailed|trace", class = "...")]` on every field. +/// - Optional struct-level or field-level uses markers for exact sink +/// selection. Field-level markers override the struct default. /// /// Event definitions inside `codex-observability` itself may use /// `#[observation(crate = "crate")]` so generated code refers to local types @@ -39,6 +43,7 @@ fn expand_observation(input: DeriveInput) -> syn::Result syn::Result syn::Result, } fn observation_attr(attrs: &[syn::Attribute]) -> syn::Result { @@ -88,6 +94,7 @@ fn observation_attr(attrs: &[syn::Attribute]) -> syn::Result { } let mut name = None; let mut crate_path = None; + let mut uses = Vec::new(); attr.parse_nested_meta(|meta| { if meta.path.is_ident("name") { name = Some(meta.value()?.parse::()?); @@ -99,6 +106,9 @@ fn observation_attr(attrs: &[syn::Attribute]) -> syn::Result { // not available. Ordinary users get the stable public path. crate_path = Some(value.parse::()?); Ok(()) + } else if meta.path.is_ident("uses") { + uses = use_literals(meta.value()?.parse::()?)?; + Ok(()) } else { Err(meta.error("unsupported observation attribute")) } @@ -107,6 +117,7 @@ fn observation_attr(attrs: &[syn::Attribute]) -> syn::Result { return Ok(ObservationAttr { name, crate_path: crate_path.unwrap_or_else(|| syn::parse_quote!(::codex_observability)), + uses, }); } } @@ -120,6 +131,7 @@ fn obs_meta( attrs: &[syn::Attribute], field_name: &syn::Ident, crate_path: &Path, + default_uses: &[LitStr], ) -> syn::Result { for attr in attrs { if !attr.path().is_ident("obs") { @@ -127,6 +139,7 @@ fn obs_meta( } let mut level = None; let mut class = None; + let mut uses = None; attr.parse_nested_meta(|meta| { if meta.path.is_ident("level") { level = Some(meta.value()?.parse::()?); @@ -134,6 +147,9 @@ fn obs_meta( } else if meta.path.is_ident("class") { class = Some(meta.value()?.parse::()?); Ok(()) + } else if meta.path.is_ident("uses") { + uses = Some(use_literals(meta.value()?.parse::()?)?); + Ok(()) } else { Err(meta.error("unsupported obs attribute")) } @@ -146,8 +162,14 @@ fn obs_meta( })?; let detail = detail_expr(&level, crate_path)?; let data_class = data_class_expr(&class, crate_path)?; + let uses = uses + .as_deref() + .unwrap_or(default_uses) + .iter() + .map(|value| field_use_expr(value, crate_path)) + .collect::>>()?; return Ok(quote! { - #crate_path::FieldMeta::new(#detail, #data_class) + #crate_path::FieldMeta::with_uses(#detail, #data_class, &[#(#uses),*]) }); } Err(syn::Error::new_spanned( @@ -156,6 +178,30 @@ fn obs_meta( )) } +fn use_literals(expr: Expr) -> syn::Result> { + let Expr::Array(array) = expr else { + return Err(syn::Error::new_spanned( + expr, + "obs uses must be a string array, for example uses = [\"analytics\"]", + )); + }; + + array + .elems + .into_iter() + .map(|elem| match elem { + Expr::Lit(ExprLit { + lit: Lit::Str(value), + .. + }) => Ok(value), + other => Err(syn::Error::new_spanned( + other, + "obs uses entries must be string literals", + )), + }) + .collect() +} + fn detail_expr(value: &LitStr, crate_path: &Path) -> syn::Result { enum_expr( value, @@ -184,6 +230,19 @@ fn data_class_expr(value: &LitStr, crate_path: &Path) -> syn::Result syn::Result { + enum_expr( + value, + "field use", + &[ + ("analytics", "Analytics"), + ("otel", "Otel"), + ("rollout_trace", "RolloutTrace"), + ], + quote!(#crate_path::FieldUse), + ) +} + fn enum_expr( value: &LitStr, label: &str, diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index cdcaba2701..899714d57d 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -15,7 +15,7 @@ pub enum InvocationType { /// Observation emitted when Codex uses an app connector during a turn. #[derive(Observation)] -#[observation(name = "app.used", crate = "crate")] +#[observation(name = "app.used", crate = "crate", uses = ["analytics"])] pub struct AppUsed<'a> { /// Model slug active for the turn where the app was used. #[obs(level = "basic", class = "operational")] diff --git a/codex-rs/observability/src/lib.rs b/codex-rs/observability/src/lib.rs index ca4ac8514e..3dfd9510ee 100644 --- a/codex-rs/observability/src/lib.rs +++ b/codex-rs/observability/src/lib.rs @@ -67,6 +67,46 @@ pub trait ObservationSink { fn observe(&self, event: &E); } +/// Visits fields intended for one sink and allowed by the supplied policy. +/// +/// This helper is the safe path for sinks that serialize fields in their +/// visitor. The explicit field-use marker selects the intended fields; the +/// policy gate then rejects unsafe metadata before the wrapped visitor can +/// materialize the value. +pub fn visit_fields_for_use( + event: &E, + field_use: FieldUse, + policy: FieldPolicy, + visitor: &mut V, +) where + E: Observation, + V: ObservationFieldVisitor, +{ + let mut visitor = PolicyFilteredVisitor { + field_use, + policy, + inner: visitor, + }; + event.visit_fields(&mut visitor); +} + +struct PolicyFilteredVisitor<'a, V> { + field_use: FieldUse, + policy: FieldPolicy, + inner: &'a mut V, +} + +impl ObservationFieldVisitor for PolicyFilteredVisitor<'_, V> +where + V: ObservationFieldVisitor, +{ + fn field(&mut self, name: &'static str, meta: FieldMeta, value: &T) { + if meta.is_used_by(self.field_use) && self.policy.allows(meta) { + self.inner.field(name, meta, value); + } + } +} + /// Policy metadata attached to a single observation field. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct FieldMeta { @@ -74,12 +114,36 @@ pub struct FieldMeta { pub detail: DetailLevel, /// Semantic/privacy class for the field. pub class: DataClass, + /// Exact sinks or projections that are intended to consume the field. + pub uses: &'static [FieldUse], } impl FieldMeta { - /// Creates metadata for a field. + /// Creates metadata for a field that is not consumed by any exact sink. pub const fn new(detail: DetailLevel, class: DataClass) -> Self { - Self { detail, class } + Self { + detail, + class, + uses: &[], + } + } + + /// Creates metadata for a field with explicit sink-use markers. + pub const fn with_uses( + detail: DetailLevel, + class: DataClass, + uses: &'static [FieldUse], + ) -> Self { + Self { + detail, + class, + uses, + } + } + + /// Returns true when the field was explicitly marked for `field_use`. + pub fn is_used_by(self, field_use: FieldUse) -> bool { + self.uses.contains(&field_use) } } @@ -144,6 +208,21 @@ pub enum DataClass { SecretRisk, } +/// Exact sink or projection that is intended to consume a field. +/// +/// This marker is separate from `DetailLevel` and `DataClass`: it expresses +/// intent, while detail/class remain guardrails that a sink policy enforces +/// before it serializes or exports the selected field. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FieldUse { + /// Remote product analytics. + Analytics, + /// OpenTelemetry events, logs, or metrics. + Otel, + /// Local rollout trace bundles. + RolloutTrace, +} + #[cfg(test)] mod tests { use super::*; @@ -156,6 +235,7 @@ mod tests { FieldMeta { detail: DetailLevel::Trace, class: DataClass::Content, + uses: &[], } ); } diff --git a/codex-rs/observability/tests/derive.rs b/codex-rs/observability/tests/derive.rs index 7506676052..6b8a0a947b 100644 --- a/codex-rs/observability/tests/derive.rs +++ b/codex-rs/observability/tests/derive.rs @@ -2,8 +2,10 @@ use codex_observability::DataClass; use codex_observability::DetailLevel; use codex_observability::FieldMeta; use codex_observability::FieldPolicy; +use codex_observability::FieldUse; use codex_observability::Observation; use codex_observability::ObservationFieldVisitor; +use codex_observability::visit_fields_for_use; use pretty_assertions::assert_eq; use serde::Serialize; use serde::Serializer; @@ -23,7 +25,7 @@ struct TurnConfigResolved<'a> { } #[derive(Observation)] -#[observation(name = "test.policy_filtered")] +#[observation(name = "test.policy_filtered", uses = ["analytics"])] struct PolicyFiltered<'a> { #[obs(level = "basic", class = "identifier")] thread_id: &'a str, @@ -36,6 +38,9 @@ struct PolicyFiltered<'a> { #[obs(level = "basic", class = "secret_risk")] api_key: PanicsIfSerialized, + + #[obs(level = "basic", class = "operational", uses = ["rollout_trace"])] + rollout_only_status: PanicsIfSerialized, } struct PanicsIfSerialized; @@ -61,11 +66,6 @@ struct CapturingVisitor { fields: Vec, } -struct PolicyCapturingVisitor { - policy: FieldPolicy, - fields: Vec, -} - impl ObservationFieldVisitor for CapturingVisitor { fn field( &mut self, @@ -81,25 +81,6 @@ impl ObservationFieldVisitor for CapturingVisitor { } } -impl ObservationFieldVisitor for PolicyCapturingVisitor { - fn field( - &mut self, - name: &'static str, - meta: FieldMeta, - value: &T, - ) { - if !self.policy.allows(meta) { - return; - } - - let value = match serde_json::to_value(value) { - Ok(value) => value, - Err(err) => panic!("allowed field should serialize: {err}"), - }; - self.fields.push(CapturedField { name, meta, value }); - } -} - #[test] fn derive_visits_annotated_fields_with_metadata() { let event = TurnConfigResolved { @@ -135,34 +116,44 @@ fn derive_visits_annotated_fields_with_metadata() { } #[test] -fn policy_visitor_does_not_serialize_denied_fields() { +fn use_and_policy_filter_before_serializing_denied_fields() { let event = PolicyFiltered { thread_id: "thread-1", status: "completed", raw_prompt: PanicsIfSerialized, api_key: PanicsIfSerialized, + rollout_only_status: PanicsIfSerialized, }; - let mut visitor = PolicyCapturingVisitor { - policy: FieldPolicy::new( + let mut visitor = CapturingVisitor::default(); + visit_fields_for_use( + &event, + FieldUse::Analytics, + FieldPolicy::new( DetailLevel::Basic, &[DataClass::Identifier, DataClass::Operational], ), - fields: Vec::new(), - }; - - event.visit_fields(&mut visitor); + &mut visitor, + ); assert_eq!( visitor.fields, vec![ CapturedField { name: "thread_id", - meta: FieldMeta::new(DetailLevel::Basic, DataClass::Identifier), + meta: FieldMeta::with_uses( + DetailLevel::Basic, + DataClass::Identifier, + &[FieldUse::Analytics], + ), value: Value::String("thread-1".to_string()), }, CapturedField { name: "status", - meta: FieldMeta::new(DetailLevel::Basic, DataClass::Operational), + meta: FieldMeta::with_uses( + DetailLevel::Basic, + DataClass::Operational, + &[FieldUse::Analytics], + ), value: Value::String("completed".to_string()), }, ] diff --git a/docs/observability-event-stream-design.md b/docs/observability-event-stream-design.md index ab84cb883a..7b743134a9 100644 --- a/docs/observability-event-stream-design.md +++ b/docs/observability-event-stream-design.md @@ -145,7 +145,7 @@ Example: use codex_observability::Observation; #[derive(Observation)] -#[observation(name = "turn.config_resolved")] +#[observation(name = "turn.config_resolved", uses = ["analytics"])] struct TurnConfigResolved<'a> { #[obs(level = "basic", class = "identifier")] thread_id: &'a str, @@ -217,25 +217,35 @@ local-only content. Each field should carry at least: +- **Use markers**: exact projections intended to consume the field, for example + `analytics`, `otel`, or `rollout_trace`. - **Detail level**: `basic`, `detailed`, or `trace`. - **Data class**: `identifier`, `operational`, `environment`, `content`, or `secret_risk`. +Use markers express intent. Detail level and data class are guardrails. A sink +must first select only fields explicitly marked for that sink, then enforce its +detail/class policy before serializing values. +Event structs may define default use markers when most fields feed the same +projection; field-level use markers override that default for mixed events. + Detail level is not privacy by itself. A tiny field can still be unsafe for remote export, and a trace-level field can be trace-level because it is large -rather than sensitive. Sinks must filter by both axes. +rather than sensitive. Data class is also not enough for selection: analytics +must not consume every basic operational field just because it would be safe. Expected sink policies: -- Analytics allows basic identifiers/operational fields plus selected - environment fields; it denies content and secret-risk fields. +- Analytics selects only fields marked `analytics`, then allows basic + identifiers/operational fields plus selected environment fields; it denies + content and secret-risk fields even if they are marked accidentally. - Rollout trace allows rich local fields, with explicit redaction rules for secret-risk material. - OTEL logs preserve today's log export behavior, including account/email only where today's policy allows them. - OTEL trace-safe events prefer lengths, counts, status, timing, and coarse categories over content. -- OTEL metrics read only the dimensions needed for existing metric names/tags. +- OTEL metrics select exact metric dimensions and then apply the OTEL policy. - Feedback upload applies an explicit user-approved policy over the ringbuffer. ## Event Taxonomy @@ -333,8 +343,9 @@ same E2E Codex run assert exact equality after stable JSON normalization ``` -At least one conformance path should apply the analytics field policy so bad -or missing annotations fail tests. Recommended first scenarios: +At least one conformance path should apply exact analytics use markers plus the +analytics guardrail policy, so missing markers and unsafe annotations fail +tests. Recommended first scenarios: - thread start - normal turn